Expert options

embark/dashboard/urls.py

@@ -10,6 +10,7 @@
 # view routing
 urlpatterns = [
     path('', views.main_dashboard, name='embark-MainDashboard'),
+    path('dashboard/', views.main_dashboard, name='embark-MainDashboard'),
     path('dashboard/main/', views.main_dashboard, name='embark-MainDashboard'),
     path('dashboard/service/', views.service_dashboard, name='embark-dashboard-service'),
     path('dashboard/report/', views.report_dashboard, name='embark-ReportDashboard'),

embark/dashboard/views.py

@@ -15,7 +15,7 @@
 from django.views.decorators.http import require_http_methods
 from django.contrib import messages
 from django.shortcuts import redirect
-from embark.helper import user_is_staff
+from embark.helper import user_is_auth
 from tracker.forms import AssociateForm
 from uploader.boundedexecutor import BoundedExecutor
 from uploader.forms import LabelForm
@@ -34,12 +34,10 @@
 @require_http_methods(["GET"])
 @login_required(login_url='/' + settings.LOGIN_URL)
 def main_dashboard(request):
-    if request.user.is_authenticated:
-        if FirmwareAnalysis.objects.filter(finished=True, failed=False).count() > 0 and Result.objects.filter(restricted=False).count() > 0:
-            return render(request, 'dashboard/mainDashboard.html', {'nav_switch': True, 'username': request.user.username})
-        messages.info(request, "Redirected - There are no Results to display yet")
-        return redirect('embark-uploader-home')
-    return HttpResponseForbidden
+    if FirmwareAnalysis.objects.filter(finished=True, failed=False).count() > 0 and Result.objects.filter(restricted=False).count() > 0:
+        return render(request, 'dashboard/mainDashboard.html', {'nav_switch': True, 'username': request.user.username})
+    messages.info(request, "Redirected - There are no Results to display yet")
+    return redirect('embark-uploader-home')
 
 
 @permission_required("users.dashboard_permission_advanced", login_url='/')
@@ -61,7 +59,7 @@ def stop_analysis(request):
         analysis = form.cleaned_data['analysis']
         analysis_object_ = FirmwareAnalysis.objects.get(id=analysis.id)
         # check if user auth
-        if request.user != analysis_object_.user:
+        if not user_is_auth(request.user, analysis_object_.user):
             return HttpResponseForbidden("You are not authorized!")
         logger.info("Stopping analysis with id %s", analysis_object_.id)
         pid = analysis_object_.pid
@@ -142,7 +140,7 @@ def show_log(request, analysis_id):
     logger.info("showing log for analyze_id: %s", analysis_id)
     firmware = FirmwareAnalysis.objects.get(id=analysis_id)
     # check if user auth TODO change to group auth
-    if request.user != firmware.user or not user_is_staff(request.user):
+    if not user_is_auth(request.user, firmware.user):
         return HttpResponseForbidden("You are not authorized!")
     # get the file path
     log_file_path_ = f"{Path(firmware.path_to_logs).parent}/emba_run.log"
@@ -169,7 +167,7 @@ def show_logviewer(request, analysis_id):
     logger.info("showing log viewer for analyze_id: %s", analysis_id)
     firmware = FirmwareAnalysis.objects.get(id=analysis_id)
     # check if user auth
-    if request.user != firmware.user or not user_is_staff(request.user):
+    if not user_is_auth(request.user, firmware.user):
         return HttpResponseForbidden("You are not authorized!")
     # get the file path
     log_file_path_ = f"{Path(firmware.path_to_logs).parent}/emba_run.log"
@@ -194,7 +192,7 @@ def delete_analysis(request, analysis_id):
     logger.info("Deleting analyze_id: %s", analysis_id)
     analysis = FirmwareAnalysis.objects.get(id=analysis_id)
     # check that the user is authorized
-    if request.user == analysis.user or request.user.is_superuser:
+    if user_is_auth(request.user, analysis.user):
         if analysis.finished is False:
             try:
                 BoundedExecutor.submit_kill(analysis.id)
@@ -229,7 +227,7 @@ def archive_analysis(request, analysis_id):
     logger.info("Archiving Analysis with id: %s", analysis_id)
     analysis = FirmwareAnalysis.objects.get(id=analysis_id)
     # check if user auth
-    if request.user != analysis.user and not request.user.is_superuser:
+    if not user_is_auth(request.user, analysis.user):
         return HttpResponseForbidden("You are not authorized!")
     if analysis.zip_file is None:
         # make archive for uuid
@@ -252,7 +250,7 @@ def hide_analysis(request, analysis_id):
     logger.info("Hiding Analysis with id: %s", analysis_id)
     analysis = FirmwareAnalysis.objects.get(id=analysis_id)
     # check if user auth
-    if request.user != analysis.user and not request.user.is_superuser:
+    if not user_is_auth(request.user, analysis.user):
         return HttpResponseForbidden("You are not authorized!")
     analysis.hidden = True
     analysis.save(update_fields=["hidden"])
@@ -290,6 +288,10 @@ def add_label(request, analysis_id):
         logger.info("User %s tryied to add label %s", request.user.username, new_label.label_name)
         # get analysis obj
         analysis = FirmwareAnalysis.objects.get(id=analysis_id)
+        # check auth
+        if not user_is_auth(request.user, analysis.user):
+            messages.error(request, 'No permissions for this analysis')
+            return redirect('..')
         analysis.label.add(new_label)
         analysis.save()
         messages.info(request, 'adding successful of ' + str(new_label))
@@ -310,6 +312,10 @@ def rm_label(request, analysis_id, label_name):
     analysis = FirmwareAnalysis.objects.get(id=analysis_id)
     # get lobel obj
     label_obj = Label.objects.get(label_name=label_name)
+    # check auth
+    if not user_is_auth(request.user, analysis.user):
+        messages.error(request, 'Removing Label failed, no permissions')
+        return redirect('..')
     analysis.label.remove(label_obj)
     analysis.save()
     messages.info(request, 'removing successful of ' + str(label_name))

embark/embark/helper.py

@@ -63,21 +63,48 @@ def cleanup_charfield(charfield) -> str:
     return charfield
 
 
-def count_emba_modules(emba_dir_path):
-    s_module_cnt, p_module_cnt, q_module_cnt, l_module_cnt, f_module_cnt, d_module_cnt = 0, 0, 0, 0, 0, 0
+def get_emba_modules(emba_dir_path) -> dict:
+    """
+    {
+        S_Modules: [
+            ('s02', 'S02_UEFI_FwHunt'),
+            ...
+        ],
+        P_modules : [...]
+    }
+    """
+    module_dict = {
+        "S_Modules": [],
+        "P_Modules": [],
+        "Q_Modules": [],
+        "L_Modules": [],
+        "F_Modules": [],
+        "D_Modules": [],
+    }
     for mod_file_ in os.listdir(f"{emba_dir_path}/modules"):
-        if mod_file_.startswith('S'):
-            s_module_cnt += 1
-        elif mod_file_.startswith('P'):
-            p_module_cnt += 1
-        elif mod_file_.startswith('F'):
-            f_module_cnt += 1
-        elif mod_file_.startswith('L'):
-            l_module_cnt += 1
-        elif mod_file_.startswith('Q'):
-            q_module_cnt += 1
-        elif mod_file_.startswith('D'):
-            d_module_cnt += 1
+        if os.path.isfile(os.path.join(f"{emba_dir_path}/modules", mod_file_)):
+            if mod_file_.startswith('S'):
+                module_dict["S_Modules"].append((str(mod_file_.split("_", 1)[0]).lower(), str(mod_file_)[:-3]))
+            elif mod_file_.startswith('P'):
+                module_dict["P_Modules"].append((str(mod_file_.split("_", 1)[0]).lower(), str(mod_file_)[:-3]))
+            elif mod_file_.startswith('F'):
+                module_dict["F_Modules"].append((str(mod_file_.split("_", 1)[0]).lower(), str(mod_file_)[:-3]))
+            elif mod_file_.startswith('L'):
+                module_dict["L_Modules"].append((str(mod_file_.split("_", 1)[0]).lower(), str(mod_file_)[:-3]))
+            elif mod_file_.startswith('Q'):
+                module_dict["Q_Modules"].append((str(mod_file_.split("_", 1)[0]).lower(), str(mod_file_)[:-3]))
+            elif mod_file_.startswith('D'):
+                module_dict["D_Modules"].append((str(mod_file_.split("_", 1)[0]).lower(), str(mod_file_)[:-3]))
+    return module_dict
+
+
+def count_emba_modules(module_dict):
+    s_module_cnt = len(module_dict["S_Modules"])
+    p_module_cnt = len(module_dict["P_Modules"])
+    q_module_cnt = len(module_dict["Q_Modules"])
+    l_module_cnt = len(module_dict["L_Modules"])
+    f_module_cnt = len(module_dict["F_Modules"])
+    d_module_cnt = len(module_dict["D_Modules"])
     return s_module_cnt, p_module_cnt, q_module_cnt, l_module_cnt, f_module_cnt, d_module_cnt
 
 
@@ -117,12 +144,22 @@ def get_version_strings():
     return embark_version, emba_version, stable_emba_version, container_version, nvd_version, github_emba_version
 
 
-def user_is_staff(user):
-    return user.is_staff
+def user_is_auth(req_user, own_user):
+    if req_user.is_superuser:
+        return True
+    elif req_user.is_staff:
+        return True
+    elif req_user.team == own_user.team:
+        return True
+    elif req_user.groups.filter(name='Administration_Group').exists() and own_user.team is None:
+        return True
+    return False
 
 
 if __name__ == '__main__':
+    import pprint
     TEST_STRING = 'Linux / v2.6.33.2'
     print(cleanup_charfield(TEST_STRING))
-
-    print(count_emba_modules(emba_dir_path="/home/cylox/embark/emba"))
+    emba_modle_list = get_emba_modules(emba_dir_path="/home/cylox/embark/emba")
+    print(pprint.pformat(emba_modle_list, indent=1))
+    print(count_emba_modules(emba_modle_list))

embark/embark/logreader.py

@@ -104,10 +104,10 @@ def save_status(self):
     @staticmethod
     def phase_identify(status_message):
         # phase patterns to match
-        pre_checker_phase_pattern = "Pre-checking phase"
-        testing_phase_pattern = "Testing phase"
-        simulation_phase_pattern = "System emulation phase"
-        reporting_phase_pattern = "Reporting phase"
+        pre_checker_phase_pattern = "Pre-checking phase"        # P-modules
+        testing_phase_pattern = "Testing phase"                 # S-Modules
+        simulation_phase_pattern = "System emulation phase"     # L-Modules
+        reporting_phase_pattern = "Reporting phase"             # P-Modules
         done_pattern = "Test ended on"
         failed_pattern = "EMBA failed in docker mode!"
 
@@ -164,6 +164,9 @@ def update_status(self, stream_item_list):
         # ignore all Q-modules for percentage calc
         if not re.match(".*Q[0-9][0-9]", stream_item_list[0]):
             self.status_msg["percentage"] = percentage
+        # ignore all D-modules for percentage calc
+        elif not re.match(".*D[0-9][0-9]", stream_item_list[0]):
+            self.status_msg["percentage"] = percentage
 
         # get copy of the current status message
         self.save_status()
@@ -238,7 +241,6 @@ def process_line(cls, inp, pat):
         """
         if re.match(pat, inp):
             return True
-        # else:
         return False
 
     def copy_file_content(self, diff):

embark/embark/settings/dev.py

@@ -8,7 +8,7 @@
 
 from dotenv import load_dotenv
 
-from embark.helper import count_emba_modules, get_version_strings
+from embark.helper import get_emba_modules, count_emba_modules, get_version_strings
 
 # Build paths inside the project like this: BASE_DIR / 'subdir'.
 BASE_DIR = Path(__file__).resolve().parent.parent.parent
@@ -274,8 +274,106 @@
 TEMP_DIR = Path("/tmp/")
 
 try:
-    EMBA_S_MOD_CNT, EMBA_P_MOD_CNT, EMBA_Q_MOD_CNT, EMBA_L_MOD_CNT, EMBA_F_MOD_CNT, EMBA_D_MOD_CNT = count_emba_modules(EMBA_ROOT)
+    EMBA_MODULE_DICT = get_emba_modules(EMBA_ROOT)
 except FileNotFoundError as file_error:
-    EMBA_S_MOD_CNT, EMBA_P_MOD_CNT, EMBA_Q_MOD_CNT, EMBA_L_MOD_CNT, EMBA_F_MOD_CNT, EMBA_D_MOD_CNT = 46, 20, 1, 10, 6, 3
+    EMBA_MODULE_DICT = {
+        'D_Modules': [
+            ('d10', 'D10_firmware_diffing'),
+            ('d02', 'D02_firmware_diffing_bin_details'),
+            ('d05', 'D05_firmware_diffing_extractor')
+        ],
+        'F_Modules': [
+            ('f02', 'F02_toolchain'),
+            ('f50', 'F50_base_aggregator'),
+            ('f15', 'F15_cyclonedx_sbom'),
+            ('f05', 'F05_qs_resolver'),
+            ('f10', 'F10_license_summary'),
+            ('f20', 'F20_vul_aggregator')
+        ],
+        'L_Modules': [
+            ('l99', 'L99_cleanup'),
+            ('l35', 'L35_metasploit_check'),
+            ('l10', 'L10_system_emulation'),
+            ('l23', 'L23_vnc_checks'),
+            ('l25', 'L25_web_checks'),
+            ('l20', 'L20_snmp_checks'),
+            ('l22', 'L22_upnp_hnap_checks'),
+            ('l15', 'L15_emulated_checks_nmap')
+        ],
+        'P_Modules': [
+            ('p15', 'P15_ubi_extractor'),
+            ('p60', 'P60_deep_extractor'),
+            ('p02', 'P02_firmware_bin_file_check'),
+            ('p35', 'P35_UEFI_extractor'),
+            ('p14', 'P14_ext_mounter'),
+            ('p07', 'P07_windows_exe_extract'),
+            ('p25', 'P25_android_ota'),
+            ('p18', 'P18_BMC_decryptor'),
+            ('p99', 'P99_prepare_analyzer'),
+            ('p50', 'P50_binwalk_extractor'),
+            ('p20', 'P20_foscam_decryptor'),
+            ('p40', 'P40_DJI_extractor'),
+            ('p22', 'P22_Zyxel_zip_decrypt'),
+            ('p17', 'P17_gpg_decompress'),
+            ('p65', 'P65_package_extractor'),
+            ('p21', 'P21_buffalo_decryptor'),
+            ('p19', 'P19_bsd_ufs_mounter'),
+            ('p23', 'P23_qemu_qcow_mounter'),
+            ('p55', 'P55_unblob_extractor'),
+            ('p10', 'P10_vmdk_extractor')
+        ],
+        'Q_Modules': [('q02', 'Q02_openai_question')],
+        'S_Modules': [
+            ('s100', 'S100_command_inj_check'),
+            ('s99', 'S99_grepit'),
+            ('s90', 'S90_mail_check'),
+            ('s03', 'S03_firmware_bin_base_analyzer'),
+            ('s20', 'S20_shell_check'),
+            ('s02', 'S02_UEFI_FwHunt'),
+            ('s45', 'S45_pass_file_check'),
+            ('s12', 'S12_binary_protection'),
+            ('s23', 'S23_lua_check'),
+            ('s110', 'S110_yara_check'),
+            ('s60', 'S60_cert_file_check'),
+            ('s35', 'S35_http_file_check'),
+            ('s24', 'S24_kernel_bin_identifier'),
+            ('s16', 'S16_ghidra_decompile_checks'),
+            ('s50', 'S50_authentication_check'),
+            ('s108', 'S108_stacs_password_search'),
+            ('s21', 'S21_python_check'),
+            ('s109', 'S109_jtr_local_pw_cracking'),
+            ('s17', 'S17_cwe_checker'),
+            ('s25', 'S25_kernel_check'),
+            ('s09', 'S09_firmware_base_version_check'),
+            ('s65', 'S65_config_file_check'),
+            ('s18', 'S18_capa_checker'),
+            ('s36', 'S36_lighttpd'),
+            ('s05', 'S05_firmware_details'),
+            ('s115', 'S115_usermode_emulator'),
+            ('s55', 'S55_history_file_check'),
+            ('s27', 'S27_perl_check'),
+            ('s80', 'S80_cronjob_check'),
+            ('s19', 'S19_apk_check'),
+            ('s95', 'S95_interesting_files_check'),
+            ('s75', 'S75_network_check'),
+            ('s106', 'S106_deep_key_search'),
+            ('s107', 'S107_deep_password_search'),
+            ('s15', 'S15_radare_decompile_checks'),
+            ('s07', 'S07_bootloader_check'),
+            ('s22', 'S22_php_check'),
+            ('s26', 'S26_kernel_vuln_verifier'),
+            ('s85', 'S85_ssh_check'),
+            ('s10', 'S10_binaries_basic_check'),
+            ('s13', 'S13_weak_func_check'),
+            ('s08', 'S08_main_package_sbom'),
+            ('s40', 'S40_weak_perm_check'),
+            ('s118', 'S118_busybox_verifier'),
+            ('s14', 'S14_weak_func_radare_check'),
+            ('s116', 'S116_qemu_version_detection'),
+            ('s04', 'S04_windows_basic_analysis'),
+            ('s06', 'S06_distribution_identification')
+        ]
+    }
+EMBA_S_MOD_CNT, EMBA_P_MOD_CNT, EMBA_Q_MOD_CNT, EMBA_L_MOD_CNT, EMBA_F_MOD_CNT, EMBA_D_MOD_CNT = count_emba_modules(EMBA_MODULE_DICT)
 
 VERSION = get_version_strings()