Merge pull request abrignoni#669 from mobasi-team/security/nq-vault-path-traversal-fix

security: block NQ Vault path traversal on decrypted file writes

Author: stark4n6

admin/test/scripts/test_nq_vault_path_security.py

@@ -0,0 +1,77 @@
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+import uuid
+
+
+ROOT_DIR = os.path.dirname(
+    os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+)
+if ROOT_DIR not in sys.path:
+    sys.path.insert(0, ROOT_DIR)
+
+from scripts.artifacts import NQ_Vault
+import scripts.ilapfuncs as ilapfuncs
+
+
+class TestNQVaultPathSecurity(unittest.TestCase):
+    def test_sanitize_output_filename_removes_traversal(self):
+        sanitized = NQ_Vault.sanitize_output_filename('../../../Users/examiner/.zshrc')
+        self.assertEqual(sanitized, 'zshrc')
+
+    def test_build_safe_output_path_is_confined_to_report_folder(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            report_folder = Path(tmpdir) / 'report'
+            report_folder.mkdir()
+            safe_path = NQ_Vault.build_safe_output_path(
+                str(report_folder),
+                '../../../../../../outside_written.bin',
+            )
+            self.assertTrue(str(safe_path).startswith(str(report_folder.resolve())))
+            self.assertEqual(safe_path.name, 'outside_written.bin')
+
+    def test_file_decryption_blocks_path_traversal_write(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            base = Path(tmpdir)
+            report_folder = base / 'report'
+            report_folder.mkdir()
+            ilapfuncs.OutputParameters.screen_output_file_path = str(base / 'screen_output.html')
+
+            image_dir = base / 'payload' / '.image'
+            image_dir.mkdir(parents=True)
+            encrypted_file = image_dir / '1700000000.bin'
+            encrypted_file.write_bytes(b'A' * 200)
+
+            unique_name = f'outside_written_{uuid.uuid4().hex}.bin'
+            traversal_name = f'../../../{unique_name}'
+            file_info = {
+                '1700000000.bin': {
+                    'old_filepath': '/sdcard/DCIM/Camera/original.jpg',
+                    'old_filename': traversal_name,
+                    'vault_filepath': '/sdcard/SystemAndroid/Data/hash/.image/1700000000.bin',
+                    'timestamp': '2024-01-01 00:00:00',
+                    'vid_length': '',
+                    'resolution': '',
+                    'alb_name': 'Default',
+                    'prev_alb_name': '',
+                    'password_id': 'enc-pin-id',
+                }
+            }
+            pin_info = {
+                'enc-pin-id': {
+                    'xor_key': '0x00',
+                    'pin_for_XOR_key': '0000',
+                }
+            }
+
+            NQ_Vault.file_decryption([str(encrypted_file)], file_info, pin_info, str(report_folder))
+
+            outside_target = (report_folder / traversal_name).resolve()
+            self.assertFalse(outside_target.exists())
+            self.assertTrue((report_folder / unique_name).exists())
+
+
+if __name__ == '__main__':
+    unittest.main()

scripts/artifacts/NQ_Vault.py

@@ -1,24 +1,44 @@
 import itertools
+import re
 import string
+import sqlite3
 from pathlib import Path
-from os.path import join
 
 from scripts.artifact_report import ArtifactHtmlReport
-from scripts.ilapfuncs import logfunc, tsv, timeline, is_platform_windows, media_to_html, open_sqlite_db_readonly
+from scripts.ilapfuncs import logfunc, tsv, media_to_html, open_sqlite_db_readonly
 
 
-'''def extract_PIN_from_db(file_found):
-    try:
-        connection = open_sqlite_db_readonly(file_found)
-
-        set_pointer = connection.cursor()
-        set_pointer.execute("SELECT passwd_temp FROM albumstemp")
+def sanitize_output_filename(filename, default_name='recovered_file.bin'):
+    raw_name = str(filename or '')
+    normalized = raw_name.replace('\\', '/')
+    safe_name = Path(normalized).name
+    safe_name = re.sub(r'[\x00-\x1f<>:"/\\|?*]+', '_', safe_name)
+    safe_name = safe_name.strip().strip('.')
+    if not safe_name:
+        return default_name
+    return safe_name
 
-        encrypted_password = set_pointer.fetchone()[0]
 
-        return encrypted_password, file_found
-    except:
-        return None'''
+def build_safe_output_path(report_folder, original_filename):
+    report_root = Path(report_folder).resolve()
+    sanitized_name = sanitize_output_filename(original_filename)
+    output_path = (report_root / sanitized_name).resolve()
+    try:
+        output_path.relative_to(report_root)
+    except ValueError:
+        output_path = report_root / sanitize_output_filename(None)
+
+    if output_path.exists():
+        stem = output_path.stem
+        suffix = output_path.suffix
+        index = 1
+        while True:
+            candidate = report_root / f'{stem}_{index}{suffix}'
+            if not candidate.exists():
+                output_path = candidate
+                break
+            index += 1
+    return output_path
 
 
 def extract_data_from_db(file_found):
@@ -63,7 +83,7 @@ def extract_data_from_db(file_found):
             dict_of_dicts[encrypted_filename] = new_dict
 
         return dict_of_dicts, list_of_enc_pins, file_found
-    except:
+    except (sqlite3.Error, OSError, ValueError, TypeError, AttributeError, IndexError, KeyError):
         return None
 
 
@@ -95,8 +115,7 @@ def brute_force_pin(encoded_PIN):
                 if decoded_test_value_string == encoded_PIN:
                     #logfunc(f'Decrypted PIN is: {pin}')
                     return pin
-            else:
-                pin_len += 1
+            pin_len += 1
 
         if decoded_PIN is None:
             print('Sorry. No PIN found.')
@@ -120,30 +139,6 @@ def raw_pin_to_XOR_key(pin):
 
     return hex_XOR_key
 
-
-'''def read_file_info(file_found):
-    file_match_dict = {}
-    print(f'readfileinfo: {file_found}')
-    connection = open_sqlite_db_readonly(file_found)
-
-    set_pointer = connection.cursor()
-    set_pointer.execute("SELECT file_name_from, file_path_new FROM hideimagevideo")
-
-    required_column = set_pointer.fetchall()
-
-    if len(required_column) < 1:
-        logfunc('No encrypted media present')
-        return
-    else:
-        for filename in required_column:
-            decrypted_filename = filename[0]
-            encrypted_filename = filename[1].split('/')[-1]
-
-            file_match_dict[decrypted_filename] = encrypted_filename
-
-    return file_match_dict'''
-
-
 def file_decryption(files_found, dict_of_file_info, dict_of_pin_dicts, report_folder):
     data_list = []
     logfunc('Encrypted files found! File Decryption in progress...')
@@ -187,12 +182,13 @@ def file_decryption(files_found, dict_of_file_info, dict_of_pin_dicts, report_fo
                                 byte = file_to_decrypt.read(1)
 
                             xord_bytes_decrypted = bytes(xor_list)
-                            with open(join(report_folder, decrypted_file_name), 'wb') as decryptedFile:
+                            decrypted_output_path = build_safe_output_path(report_folder, decrypted_file_name)
+                            with open(decrypted_output_path, 'wb') as decryptedFile:
                                 decryptedFile.write(xord_bytes_decrypted)
                                 decryptedFile.close()
 
                                 tolink = []
-                                pathdec = join(report_folder, decrypted_file_name)
+                                pathdec = str(decrypted_output_path)
                                 tolink.append(pathdec)
                                 thumb = media_to_html(pathdec, tolink, report_folder)
 
@@ -213,24 +209,28 @@ def file_decryption(files_found, dict_of_file_info, dict_of_pin_dicts, report_fo
                                                              html_no_escape=['Media'])
                             report.end_artifact_report()
 
-                            tsvname = f'NQVault'
+                            tsvname = 'NQVault'
                             tsv(report_folder, data_headers, data_list, tsvname)
 
 
 # MAIN #
 def get_NQVault(files_found, report_folder, seeker, wrap_text):
+    _ = seeker, wrap_text
     data_list = []
     list_of_enc_pins = []
     dict_of_file_info = {}
     dict_of_pin_dicts = {}
-    sucess = 0
+    file_found_pin = ''
+    success = 0
     # Get the "encrypted" PIN from DB
     for file_found in files_found:
         if file_found.endswith('322w465ay423xy11'):
             # multiple password hashes may be present, so these are stored as a list
-            dict_of_file_info, list_of_enc_pins, file_found_pin = extract_data_from_db(file_found)
-            sucess = 1
-    if sucess == 0:
+            extraction_result = extract_data_from_db(file_found)
+            if extraction_result:
+                dict_of_file_info, list_of_enc_pins, file_found_pin = extraction_result
+                success = 1
+    if success == 0:
         logfunc('No Database DB Found or no hashed PIN present.')
         return
 
@@ -256,7 +256,8 @@ def get_NQVault(files_found, report_folder, seeker, wrap_text):
         report.add_script()
         data_headers = ('Encrypted PIN', 'Decrypted PIN')
 
-        report.write_artifact_data_table(data_headers, data_list, file_found_pin, html_no_escape=['Media'])
+        report_source = file_found_pin if file_found_pin else 'NQ Vault Source DB'
+        report.write_artifact_data_table(data_headers, data_list, report_source, html_no_escape=['Media'])
         report.end_artifact_report()
 
         # Media decryption funct