Deep scan enchancement (Samsung#235)

* Encoded data might be decoded

* apply file type when not None

* fix

* Separated test for docx

* Encoded test

* Improved encode test research

* fix

* refuzzed

* Commited forgotten sample

* Update credsweeper/file_handler/data_content_provider.py

Co-authored-by: ShinHyung Choi <sh519.choi@samsung.com>

* Update credsweeper/app.py

Co-authored-by: ShinHyung Choi <sh519.choi@samsung.com>

* Apply test assertion

* Use common code for reduce duplicate code

* Line umeration

* fix

* make the method private again

* rename methods

Co-authored-by: ShinHyung Choi <sh519.choi@samsung.com>

credsweeper/app.py

@@ -20,6 +20,7 @@
 from credsweeper.file_handler.diff_content_provider import DiffContentProvider
 from credsweeper.file_handler.file_path_extractor import FilePathExtractor
 from credsweeper.file_handler.files_provider import FilesProvider
+from credsweeper.file_handler.string_content_provider import StringContentProvider
 from credsweeper.file_handler.text_content_provider import TextContentProvider
 from credsweeper.scanner import Scanner
 from credsweeper.utils import Util
@@ -250,19 +251,30 @@ def file_scan(self, content_provider: ContentProvider) -> List[Candidate]:
                                                             content_provider.file_type, content_provider.info)
             candidates.append(dummy_candidate)
 
-        elif self.config.depth > 0 and isinstance(content_provider, TextContentProvider):
-            # Feature to scan files which might be containers
-            data = Util.read_data(content_provider.file_path)
-            if data:
-                data_provider = DataContentProvider(data=data,
-                                                    file_path=content_provider.file_path,
-                                                    info=content_provider.file_path)
-                candidates = self.data_scan(data_provider, self.config.depth, RECURSIVE_SCAN_LIMITATION)
-
         else:
             # Regular file scanning
-            analysis_targets = content_provider.get_analysis_target()
-            candidates = self.scanner.scan(analysis_targets)
+            if content_provider.file_type not in self.config.exclude_containers:
+                analysis_targets = content_provider.get_analysis_target()
+                candidates.extend(self.scanner.scan(analysis_targets))
+
+            if self.config.depth and isinstance(content_provider, TextContentProvider):
+                # Feature to scan files which might be containers
+                data = Util.read_data(content_provider.file_path)
+                if data:
+                    data_provider = DataContentProvider(data=data,
+                                                        file_path=content_provider.file_path,
+                                                        file_type=content_provider.file_type,
+                                                        info=content_provider.file_path)
+                    extra_candidates = self.data_scan(data_provider, self.config.depth, RECURSIVE_SCAN_LIMITATION)
+                    if extra_candidates:
+                        # reduce duplicated credentials
+                        found_values = set(line_data.value for candidate in candidates
+                                           for line_data in candidate.line_data_list)
+                        for extra_candidate in extra_candidates:
+                            for line_data in extra_candidate.line_data_list:
+                                if line_data.value not in found_values:
+                                    candidates.append(extra_candidate)
+                                    break
 
         # finally return result from 'file_scan'
         return candidates
@@ -336,6 +348,22 @@ def data_scan(self, data_provider: DataContentProvider, depth: int, recursive_li
             except Exception as gzip_exc:
                 logger.error(f"{data_provider.file_path}:{gzip_exc}")
 
+        elif data_provider.represent_as_encoded():
+            decoded_data_provider = DataContentProvider(data=data_provider.decoded,
+                                                        file_path=data_provider.file_path,
+                                                        file_type=data_provider.file_type,
+                                                        info=f"{data_provider.info}|ENCODED")
+            new_limit = recursive_limit_size - len(decoded_data_provider.data)
+            candidates.extend(self.data_scan(decoded_data_provider, depth, new_limit))
+
+        elif data_provider.represent_as_xml():
+            struct_data_provider = StringContentProvider(lines=data_provider.lines,
+                                                         line_numbers=data_provider.line_numbers,
+                                                         file_path=data_provider.file_path,
+                                                         file_type=".xml",
+                                                         info=f"{data_provider.info}|XML")
+            candidates.extend(self.file_scan(struct_data_provider))
+
         else:
             # finally try scan the data via byte content provider
             byte_content_provider = ByteContentProvider(content=data_provider.data,

credsweeper/file_handler/content_provider.py

@@ -14,7 +14,7 @@ def __init__(
             file_type: Optional[str] = None,  #
             info: Optional[str] = None) -> None:
         self.file_path: str = file_path
-        self.file_type: str = file_type if file_type else Util.get_extension(file_path)
+        self.file_type: str = file_type if file_type is not None else Util.get_extension(file_path)
         self.info: str = info
 
     @abstractmethod

credsweeper/file_handler/data_content_provider.py

@@ -1,7 +1,14 @@
+import base64
+import logging
+import string
 from typing import List, Optional
 
+from credsweeper.common.constants import DEFAULT_ENCODING
 from credsweeper.file_handler.analysis_target import AnalysisTarget
 from credsweeper.file_handler.content_provider import ContentProvider
+from credsweeper.utils import Util
+
+logger = logging.getLogger(__name__)
 
 
 class DataContentProvider(ContentProvider):
@@ -21,6 +28,9 @@ def __init__(
             info: Optional[str] = None) -> None:
         super().__init__(file_path=file_path, file_type=file_type, info=info)
         self.data = data
+        self.decoded: Optional[bytes] = None
+        self.lines: List[str] = []
+        self.line_numbers: List[int] = []
 
     @property
     def data(self) -> bytes:
@@ -32,6 +42,40 @@ def data(self, data: bytes) -> None:
         """data setter"""
         self.__data = data
 
+    def represent_as_xml(self) -> bool:
+        """Tries to read data as xml
+
+        Return:
+             True if reading was successful
+
+        """
+        try:
+            xml_text = self.data.decode(encoding=DEFAULT_ENCODING).splitlines()
+            self.lines, self.line_numbers = Util.get_xml_from_lines(xml_text)
+        except Exception as exc:
+            logger.debug("Cannot parse as XML:%s %s", exc, self.data)
+            return False
+        return bool(self.lines and self.line_numbers)
+
+    def represent_as_encoded(self) -> bool:
+        """Encodes data from base64. Stores result in decoded
+
+        Return:
+             True if the data correctly parsed and verified
+
+        """
+        if len(self.data) < 12 or (b"=" in self.data and b"=" != self.data[-1]):
+            logger.debug("Weak data to decode from base64: %s", self.data)
+        try:
+            self.decoded = base64.b64decode(  #
+                self.data.decode(encoding="ascii", errors="strict").  #
+                translate(str.maketrans("", "", string.whitespace)),  #
+                validate=True)  #
+        except Exception as exc:
+            logger.debug("Cannot decoded as base64:%s %s", exc, self.data)
+            return False
+        return self.decoded is not None and 0 < len(self.decoded)
+
     def get_analysis_target(self) -> List[AnalysisTarget]:
         """Return nothing. The class provides only data storage.
 

credsweeper/file_handler/string_content_provider.py

@@ -16,11 +16,15 @@ class StringContentProvider(ContentProvider):
     def __init__(
             self,  #
             lines: List[str],  #
+            line_numbers: Optional[List[int]] = None,  #
             file_path: Optional[str] = None,  #
             file_type: Optional[str] = None,  #
             info: Optional[str] = None) -> None:
         super().__init__(file_path=file_path, file_type=file_type, info=info)
         self.lines = lines
+        # fill line numbers only when amounts are equal
+        self.line_numbers = line_numbers if line_numbers and len(self.lines) == len(line_numbers) \
+            else (list(range(1, 1 + len(self.lines))) if self.lines else [])
 
     def get_analysis_target(self) -> List[AnalysisTarget]:
         """Return lines to scan.
@@ -30,6 +34,6 @@ def get_analysis_target(self) -> List[AnalysisTarget]:
 
         """
         return [
-            AnalysisTarget(line, i + 1, self.lines, self.file_path, self.file_type, self.info)
-            for i, line in enumerate(self.lines)
+            AnalysisTarget(line, line_number, self.lines, self.file_path, self.file_type, self.info)
+            for line_number, line in zip(self.line_numbers, self.lines)
         ]

credsweeper/utils/util.py

@@ -272,6 +272,27 @@ def read_data(path: str) -> Optional[bytes]:
             logger.error(f"Unexpected Error: Can not read '{path}'. Error message: '{exc}'")
         return None
 
+    @staticmethod
+    def get_xml_from_lines(xml_lines: List[str]) -> Tuple[Optional[List[str]], Optional[List[int]]]:
+        """Parse xml data from list of string and return List of str.
+
+        Args:
+            xml_lines: list of lines of xml data
+
+        Return:
+            List of formatted string(f"{root.tag} : {root.text}")
+
+        """
+        lines = []
+        line_nums = []
+        tree = etree.fromstringlist(xml_lines)
+        for element in tree.iter():
+            tag = Util._extract_element_data(element, "tag")
+            text = Util._extract_element_data(element, "text")
+            lines.append(f"{tag} : {text}")
+            line_nums.append(element.sourceline)
+        return lines, line_nums
+
     @staticmethod
     def get_xml_data(file_path: str) -> Tuple[Optional[List[str]], Optional[List[int]]]:
         """Read xml data and return List of str.
@@ -285,21 +306,13 @@ def get_xml_data(file_path: str) -> Tuple[Optional[List[str]], Optional[List[int
             List of formatted string(f"{root.tag} : {root.text}")
 
         """
-        lines = []
-        line_nums = []
         try:
             with open(file_path, "r") as f:
                 xml_lines = f.readlines()
-            tree = etree.fromstringlist(xml_lines)
-            for element in tree.iter():
-                tag = Util._extract_element_data(element, "tag")
-                text = Util._extract_element_data(element, "text")
-                lines.append(f"{tag} : {text}")
-                line_nums.append(element.sourceline)
+            return Util.get_xml_from_lines(xml_lines)
         except Exception as exc:
             logger.error(f"Cannot parse '{file_path}' to xml {exc}")
             return None, None
-        return lines, line_nums
 
     @staticmethod
     def _extract_element_data(element, attr) -> str:

fuzz/corpus/1df555ea6ab8f834626d3002c2d3eaf7746450ae

@@ -0,0 +1 @@
+<link rel="0ERgAAQA06dded84d6a99f6126968f210a526d9bb8E8bfcZXjU//9k="/>

fuzz/corpus/d4b0733705909d11a644c3a0c59b097ac596c427 → fuzz/corpus/3b5955751f9abbb7e0d7001ccea24de9d21d9277

@@ -1,2 +1,2 @@
 AKIAIMWD4EU8OPQMSH5X
-password I3vKF0Pff8O+z+yskL9wc8ldgYkSN0mJ54MVCwb9LLkC3SL41R7vTPvzVAotOO3wb7oOLHb5rwZP8PA6Zmvm9qiPDis8PpaVMyBpczwuBY3OQkjSpNLau7Zym6zn5+69UgzvwfONTWtEdQd8Ch+L1a/BX8cT0KoOep83igh6ZD+ga8iMFQioSzlTFZMUPzmmKGabxIBNu/O5jXI9rnPgg5RarnDCNYiocUJJoCwtUC5QbpHMrZEpIkjPFW2Frh7noXtIdhJGeLTYxPO/0LwJDqtLKuJRucB8cCiHpjU1dprJmahR+PFMWofj2nD/qYhLFc4XVLcWmR4R+HoeT6C70vMR3oHJDZLulxlupactHI8yDGuoklRtptfd+O908bHNdPbzKQWxlp/kSjv7fTFHtBTgcj40rbv1gJduegl8wkN3/NUd4tDkfbS4799io/Rc+uUFprr+yUG+Fxr2x3DagitKJ6wjx9JuPFPBHIxNCt/lSAIGMnMI3XHWEuFe1NWAoGbMee8YD/EgWEcbwCfS7uvr/+G5TqaevltipBtgyv3XIWVkp/RuNArmwY5Jgub6goNsZkjJqyfD+U9L8rjLcT19PJD79epxH+jPQLC6cwb3lizYflo7tEJSsr1/N3SOKwQtH0Ly2UcZeTL5YG4I4RUdY0WWfOpA+2nZtpwaRR64yLdHg1J0P5VQoSI3zKz0JPn8a2CS9xwCZY7u4vwjPvdRYpvBaWmEOOPM6/9MaOxw3NNREabYJGqaqc4xQV81XAV7vQjK42bQpf9+gagtkVBsED8UdblWf4pIWDBHROjaeoD6JqsRRYduijIwVtkT1MsdomkG1/EGM11Aq/aKGj7sjxTzKzxYjkftxMCvjffT0dRVLpjpwSqKUv/EIPdHLMOzD9gh7mLW34SsndDB9pGjTVWkdpu8DiKIyh964MkJIUS9PSD5xIYvMymmXkvS5fGXpjwsdv4/x/RDvhAwlZA1zGyL3K7Qwp3XAx9exp04FC3Fs8XCPsg5XFdlchuntUAhD9mJ5u7enLDVZGHtoCUS3z8u+OHjNuyFTZNMtGt1zX07EbiD8K5IrPDeNbTSDtBa6tnEnIFHY3q5kBSstjAwIRkC512L8XZX3fNBeMFELhc4Ms2YJhkcFnrSYehfjn6Av8x+BpZeZMS6Q4jGiD6juXHsViZv21e4FoX5SKbfQiOMAPCSNkxz5j/RJTid+8WAekXQ1SyPwzt81oQZbU2rmNTJ65Bk3NSCteCkA+VsyoCw5g/GMlakZ6jPPraAwHPA9f7l+ByZizPW7XzP/MwGryzHhXqjDLwqNCr/C0MotJgzTKYybbHvoVGEj70s5x7PYtMEED/1KqpRuVjBplg7oLIJVTl6+1a6U19Hx0BNfnw/Jds+Q4I2eaoVb7Lh9Ql+OFDRAvhp/JWmE1c+jZEHPikYysoV1bsBXbe682B3TFVbHtdjm7M6RlqU8e8ohWhPP0OxDIFqs56QdtR1
+password I3vKF0Pff8O+z+yskL9wc8ldgYkSN0mJ54MVCwb9LLkC3SL41R7vTPvzVAotOO3wb7oOLHb5rwZP8PA6Zmvm9qiPDis8PpaVMyBpczwuBY3OQkjSpNLau7Zym6zn5+69UgzvwfONTWtEdQd8Ch+L1a/BX8cT0KoOep83igh6ZD+ga8iMFQioSzlTFZMUPzmmKGabxIBNu/O5jXI9rnPgg5RarnDCNYiocUJJoCwtUC5QbpHMrZEpIkjPFW2Frh7noXtIdhJGeLTYxPO/0LwJDqtLKuJRucB8cCiHpjU1dprJmahR+PFMWofj2nD/qYhLFc4XVLcWmR4R+HoeT6C70vMR3oHJDZLulxlupactHI8yDGuoklRtptfd+O908bHNdPbzKQWxlp/kSjv7fTFHtBTgcj40rbv1gJduegl8wkN3/NUd4tDkfbS4799io/Rc+uUFprr+yUG+Fxr2x3DagitKJ6wjx9JuPFPBHIxNCt/lSAIGMnMI3XHWEuFe1NWAoGbMee8YD/EgWEcbwCfS7uvr/+G5TqaevltipBtgyv3XIWVkp/RuNArmwY5Jgub6goNsZkjJqyfD+U9L8rjLcT19PJD79epxH+jPQLC6cwb3lizYflo7tEJSsr1/N3SOKwQtH0Ly2UcZeTL5YG4I4RUdY0WWfOpA+2nZtpwaRR64yLdHg1J0P5VQoSI3zKz0JPn8a2CS9xwCZY7u4vwjPvdRYpvBaWmEOOPM6/9MaOxw3NNREabYJGqaqc4xQV81XAV7vQjK42bQpf9+gagtkVBsED8UdblWf4pIWDBHROjaeoD6JqsRRYduijIwVtkT1MsdomkG1/EGM11Aq/aKGj7sjxTzKzxYjkftxMCvjffT0dRVLpjpwSqKUvѺ��dHLMOzD9gh7mLW34SsndDB9pGjTVWkdpu8DiKIyh964MkJIUS9PSD5xIYvMymmXkvS5fGXpjwsdv4/x/RDvhAwlZA1zGyL3K7Qwp3XAx9exp04FC3Fs8XCPsg5XFdlchuntUAhD9mJ5u7enLDVZGHtoCUS3z8u+OHjNuyFTZNMtGt1zX07EbiD8K5IrPDeNbTSDtBa6tnEnIFHY3q5kBSstjAwIRkC512L8XZX3fNBeMFELhc4Ms2YJhkcFnrSYehfjn6Av8x+BpZeZMS6Q4jGiD6juXHsViZv21e4FoX5SKbfQiOMAPCSNkxz5j/RJTid+8WAekXQ1SyPwzt81oQZbU2rmNTJ65Bk3NSCteCkA+VsyoCw5g/GMlakZ6jPPraAwHPA9f7l+ByZizPW7XzP/MwGryzHhXqjDLwqNCr/C0MotJgzTKYybbHvoVGEj70s5x7PYtMEED/1KqpRuVjBplg7oLIJVTl6+1a6U19Hx0BNfnw/Jds+Q4I2eaoVb7Lh9Ql+OFDRAvhp/JWmE1c+jZEHPikYysoV1bsBXbe682B3TFVbHtdjm7M6RlqU8e8ohWhPP0OxDIFqs56QdtR1

fuzz/corpus/50127775b46b8b432f85928ab5d21cc9a5b9916a

@@ -0,0 +1,5 @@
+<?xml version��ity>
+        <password name="password">cackl/City>
+        <password name="password">peace_for_ukraine</password>
+    </Country>
+</Countries>

tests/__init__.py

@@ -1,7 +1,7 @@
 from pathlib import Path
 
 # total number of files in test samples, included .gitignore
-SAMPLES_FILES_COUNT: int = 50
+SAMPLES_FILES_COUNT: int = 52
 
 # credentials count after scan
 SAMPLES_CRED_COUNT: int = 51
@@ -11,9 +11,9 @@
 SAMPLES_POST_CRED_COUNT: int = 20
 
 # archived credentials that not found without --depth
-SAMPLES_IN_DEEP_1 = 2
-SAMPLES_IN_DEEP_2 = 3
-SAMPLES_IN_DEEP_3 = 4
+SAMPLES_IN_DEEP_1 = 6
+SAMPLES_IN_DEEP_2 = 7
+SAMPLES_IN_DEEP_3 = 8
 
 SAMPLES_FILTERED_BY_POST_COUNT = 1
 

tests/file_handler/test_data_content_provider.py

@@ -102,17 +102,18 @@ def test_scan_zipfile_p(self) -> None:
                             with open(os.path.join(dirpath, filename), "rb") as input_file:
                                 output_file.write(input_file.read())
                                 samples_file_count += 1
-            assert samples_file_count == SAMPLES_FILES_COUNT
+            self.assertEqual(SAMPLES_FILES_COUNT, samples_file_count)
             content_provider = TextProvider([zip_file_path])
             file_extractors = content_provider.get_scannable_files(cs.config)
             assert len(file_extractors) == 1
             # single extractor
             zip_scan_results = cs.file_scan(file_extractors[0])
+            self.assertEqual(len_samples_scan_results, len(zip_scan_results))
+
             cs.credential_manager.set_credentials(zip_scan_results)
             cs.post_processing()
             cs.export_results()
 
-            assert len_samples_scan_results == len(zip_scan_results)
             assert os.path.isfile(report_path_1)
             with open(report_path_1) as f:
                 report = json.load(f)

tests/file_handler/test_string_content_provider.py

@@ -1,22 +1,47 @@
-from typing import List
-
-import pytest
-
 from credsweeper.file_handler.analysis_target import AnalysisTarget
 from credsweeper.file_handler.string_content_provider import StringContentProvider
 
 
 class TestStringContentProvider:
 
-    @pytest.mark.parametrize("lines", [["line one", "password='in_line_2'"]])
-    def test_get_analysis_target_p(self, lines: List[str]) -> None:
+    def test_get_analysis_target_p(self) -> None:
         """Evaluate that lines data correctly extracted from file"""
+        lines = ["line one", "password='in_line_2'"]
         content_provider = StringContentProvider(lines)
         analysis_targets = content_provider.get_analysis_target()
 
+        assert len(analysis_targets) == len(lines)
+
         expected_target = AnalysisTarget(lines[0], 1, lines, "", "", "")
+        assert analysis_targets[0] == expected_target
+        # check second target and line numeration
+        expected_target = AnalysisTarget(lines[1], 2, lines, "", "", "")
+        assert analysis_targets[1] == expected_target
 
-        assert len(analysis_targets) == len(lines)
+        # specific line numeration
+        content_provider = StringContentProvider(lines, [42, -1])
+        analysis_targets = content_provider.get_analysis_target()
+        assert analysis_targets[0].line_num == 42
+        assert analysis_targets[1].line_num == -1
 
-        target = analysis_targets[0]
-        assert target == expected_target
+    def test_get_analysis_target_n(self) -> None:
+        """Negative cases check"""
+        # empty list
+        content_provider = StringContentProvider([])
+        analysis_targets = content_provider.get_analysis_target()
+        assert len(analysis_targets) == 0
+
+        # mismatched amount of lists
+        content_provider = StringContentProvider(["a", "b", "c"], [2, 3])
+        analysis_targets = content_provider.get_analysis_target()
+        assert len(analysis_targets) == 3
+        assert analysis_targets[0].line_num == 1
+        assert analysis_targets[1].line_num == 2
+        assert analysis_targets[2].line_num == 3
+
+        content_provider = StringContentProvider(["a", "b", "c"], [5, 3, 4, 5])
+        analysis_targets = content_provider.get_analysis_target()
+        assert len(analysis_targets) == 3
+        assert analysis_targets[0].line_num == 1
+        assert analysis_targets[1].line_num == 2
+        assert analysis_targets[2].line_num == 3

tests/samples/encoded

@@ -0,0 +1,7 @@
+
+XG5naXRfdG9rZW4gPSAiZ2lyZW9naWNyYWNrbGVjcmFja2xlMTIzMTU2NzE5MDEx
+
+         MzQxMzk4MSJc
+
+blxuCg==
+