test: add missing analyzer unit tests and datafinder date validation

timesketch/lib/analyzers/domain_test.py

@@ -1,3 +1,16 @@
+# Copyright 2019 Google Inc. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
 """Tests for DomainPlugin."""
 
 from unittest import mock
@@ -19,3 +32,86 @@ def test_domain_analyzer_class(self):
         analyzer = domain.DomainSketchPlugin(index_name, sketch_id)
         self.assertEqual(analyzer.index_name, index_name)
         self.assertEqual(analyzer.sketch.id, sketch_id)
+
+    @mock.patch("timesketch.lib.analyzers.interface.OpenSearchDataStore", MockDataStore)
+    def test_domain_analyzer_no_events(self):
+        """Test that the analyzer handles an empty event stream gracefully."""
+        analyzer = domain.DomainSketchPlugin("test", 1, timeline_id=1)
+        analyzer.datastore.client = mock.Mock()
+
+        result = analyzer.run()
+        self.assertIn("No domains to analyze.", result)
+
+    @mock.patch("timesketch.lib.analyzers.interface.OpenSearchDataStore", MockDataStore)
+    def test_domain_extracted_from_url(self):
+        """Test that domains are extracted from the url field when domain is absent."""
+        analyzer = domain.DomainSketchPlugin("test", 1, timeline_id=1)
+        analyzer.datastore.client = mock.Mock()
+
+        # MockDataStore.search_stream iterates with str(i) keys.
+        analyzer.datastore.event_store["0"] = {
+            "_id": "0",
+            "_index": "test",
+            "_source": {
+                "__ts_timeline_id": 1,
+                "timestamp": 1410593222543942,
+                "url": "https://www.example.com/some/path?q=1",
+            },
+        }
+
+        result = analyzer.run()
+        self.assertIn("1 domains discovered", result)
+
+    @mock.patch("timesketch.lib.analyzers.interface.OpenSearchDataStore", MockDataStore)
+    def test_rare_domain_tagged(self):
+        """Test that domains appearing infrequently are counted in the result."""
+        analyzer = domain.DomainSketchPlugin("test", 1, timeline_id=1)
+        analyzer.datastore.client = mock.Mock()
+
+        # Populate the datastore: one common domain (many hits) and one
+        # rare domain (a single hit so it falls below the 20th percentile).
+        # Keys must be sequential str(i) for MockDataStore.search_stream.
+        for i in range(10):
+            analyzer.datastore.event_store[str(i)] = {
+                "_id": str(i),
+                "_index": "test",
+                "_source": {
+                    "__ts_timeline_id": 1,
+                    "timestamp": 1410593222543942 + i,
+                    "domain": "common.example.com",
+                },
+            }
+
+        analyzer.datastore.event_store["10"] = {
+            "_id": "10",
+            "_index": "test",
+            "_source": {
+                "__ts_timeline_id": 1,
+                "timestamp": 1410593222543942,
+                "domain": "rare.example.net",
+            },
+        }
+
+        result = analyzer.run()
+        self.assertIn("2 domains discovered", result)
+
+    @mock.patch("timesketch.lib.analyzers.interface.OpenSearchDataStore", MockDataStore)
+    def test_cdn_domain_tagged(self):
+        """Test that events whose domain is a known CDN are reported."""
+        analyzer = domain.DomainSketchPlugin("test", 1, timeline_id=1)
+        analyzer.datastore.client = mock.Mock()
+
+        # Keys must be sequential str(i) for MockDataStore.search_stream.
+        analyzer.datastore.event_store["0"] = {
+            "_id": "0",
+            "_index": "test",
+            "_source": {
+                "__ts_timeline_id": 1,
+                "timestamp": 1410593222543942,
+                # .cloudfront.net is listed in KNOWN_CDN_DOMAINS as Amazon CloudFront.
+                "domain": "assets.example.cloudfront.net",
+            },
+        }
+
+        result = analyzer.run()
+        self.assertIn("1 known CDN networks found.", result)

timesketch/lib/analyzers/feature_extraction_test.py

@@ -197,7 +197,36 @@ def _config_validation(self, config):
         if aggregate:
             self.assertIsInstance(aggregate, bool)
 
-    # TODO: Add tests for the feature extraction.
+    # Mock the OpenSearch datastore.
+    @mock.patch("timesketch.lib.analyzers.interface.OpenSearchDataStore", MockDataStore)
+    def test_analyzer_properties(self):
+        """Test getters and setters of FeatureExtractionSketchPlugin."""
+        analyzer = FeatureExtractionSketchPlugin("test_index", 1, 1)
+        analyzer.plugin_name = "test_plugin"
+        analyzer.feature_name = "test_feature"
+        analyzer.feature_config = {"test": "config"}
+
+        self.assertEqual(analyzer.plugin_name, "test_plugin")
+        self.assertEqual(analyzer.feature_name, "test_feature")
+        self.assertEqual(analyzer.feature_config, {"test": "config"})
+
+    @mock.patch("timesketch.lib.analyzers.interface.OpenSearchDataStore", MockDataStore)
+    def test_run_empty_plugin_name(self):
+        """Test analyzer run method with empty plugin name."""
+        analyzer = FeatureExtractionSketchPlugin("test_index", 1, 1)
+        # plugin_name is None by default
+        result = analyzer.run()
+        self.assertEqual(result, "Feature extraction plugin name is empty")
+
+    @mock.patch("timesketch.lib.analyzers.interface.OpenSearchDataStore", MockDataStore)
+    def test_run_unregistered_plugin(self):
+        """Test analyzer run method with an unregistered plugin."""
+        analyzer = FeatureExtractionSketchPlugin("test_index", 1, 1)
+        analyzer.plugin_name = "unregistered_plugin"
+        # Since it's unregistered, the PluginManager will return None
+        result = analyzer.run()
+        self.assertTrue(result.startswith("Error: Feature extraction plugin"))
+
     def test_config(self):
         """Tests that the config file is valid."""
         config_file = os.path.join("data", "regex_features.yaml")

timesketch/lib/analyzers/ntfs_timestomp_test.py

@@ -108,9 +108,85 @@ def test_is_suspicious(self):
             self.assertEqual(std_diffs, tc.expected_si_diffs)
             self.assertEqual(fn_diffs, tc.expected_fn_diffs)
 
-    # Mock the OpenSearch datastore.
     @mock.patch("timesketch.lib.analyzers.interface.OpenSearchDataStore", MockDataStore)
-    def test_analyzer(self):
-        """Test analyzer."""
-        # TODO: Write actual tests here.
-        self.assertEqual(True, True)
+    def test_is_suspicious_no_std_info(self):
+        """Test that FileInfo without a std_info_event is never suspicious."""
+        analyzer = ntfs_timestomp.NtfsTimestompSketchPlugin("test", 1)
+        file_info = ntfs_timestomp.FileInfo(
+            file_reference=1,
+            timestamp_desc="Content Modification Time",
+            std_info_event=None,
+            std_info_timestamp=0,
+            file_names=[(MockEvent(), 9000000000)],
+        )
+        self.assertFalse(analyzer.is_suspicious(file_info))
+
+    @mock.patch("timesketch.lib.analyzers.interface.OpenSearchDataStore", MockDataStore)
+    def test_is_suspicious_no_file_names(self):
+        """Test that FileInfo with no file_names is never suspicious."""
+        analyzer = ntfs_timestomp.NtfsTimestompSketchPlugin("test", 1)
+        file_info = ntfs_timestomp.FileInfo(
+            file_reference=1,
+            timestamp_desc="Content Modification Time",
+            std_info_event=MockEvent(),
+            std_info_timestamp=0,
+            file_names=[],
+        )
+        self.assertFalse(analyzer.is_suspicious(file_info))
+
+    @mock.patch("timesketch.lib.analyzers.interface.OpenSearchDataStore", MockDataStore)
+    def test_is_suspicious_within_threshold(self):
+        """Test that file_names within the threshold are not flagged."""
+        analyzer = ntfs_timestomp.NtfsTimestompSketchPlugin("test", 1)
+        # threshold is 10 * 60000000 = 600000000 microseconds by default.
+        # A diff smaller than the threshold must not trigger detection.
+        fn_event = MockEvent()
+        file_info = ntfs_timestomp.FileInfo(
+            file_reference=1,
+            timestamp_desc="Content Modification Time",
+            std_info_event=MockEvent(),
+            std_info_timestamp=0,
+            file_names=[(fn_event, analyzer.threshold - 1)],
+        )
+        self.assertFalse(analyzer.is_suspicious(file_info))
+        # The time_delta attribute must NOT have been set on the file_name event.
+        self.assertIsNone(fn_event.source.get("time_delta"))
+
+    @mock.patch("timesketch.lib.analyzers.interface.OpenSearchDataStore", MockDataStore)
+    def test_is_suspicious_timestomped(self):
+        """Test that file_names with differences above the threshold are flagged."""
+        analyzer = ntfs_timestomp.NtfsTimestompSketchPlugin("test", 1)
+        fn_event = MockEvent()
+        si_event = MockEvent()
+        large_diff = analyzer.threshold + 1
+        file_info = ntfs_timestomp.FileInfo(
+            file_reference=1,
+            timestamp_desc="Content Modification Time",
+            std_info_event=si_event,
+            std_info_timestamp=0,
+            file_names=[(fn_event, large_diff)],
+        )
+        self.assertTrue(analyzer.is_suspicious(file_info))
+        # The time_delta attribute must be set on every suspicious file_name.
+        self.assertEqual(fn_event.source.get("time_delta"), large_diff)
+        # The accumulated time_deltas must be stored on the STD_INFO event.
+        self.assertEqual(si_event.source.get("time_deltas"), [large_diff])
+
+    @mock.patch("timesketch.lib.analyzers.interface.OpenSearchDataStore", MockDataStore)
+    def test_is_suspicious_mixed_file_names(self):
+        """Test that a single within-threshold file_name prevents detection."""
+        analyzer = ntfs_timestomp.NtfsTimestompSketchPlugin("test", 1)
+        fn_event_large = MockEvent()
+        fn_event_small = MockEvent()
+        file_info = ntfs_timestomp.FileInfo(
+            file_reference=1,
+            timestamp_desc="Content Modification Time",
+            std_info_event=MockEvent(),
+            std_info_timestamp=0,
+            file_names=[
+                (fn_event_large, analyzer.threshold + 1),
+                (fn_event_small, analyzer.threshold - 1),
+            ],
+        )
+        # One file_name is within threshold — detection must be suppressed.
+        self.assertFalse(analyzer.is_suspicious(file_info))

timesketch/lib/datafinder.py

@@ -14,13 +14,30 @@
 """The class definitions for the data finder, or the data analyzer."""
 
 import logging
+from datetime import datetime
 
 from timesketch.lib.analyzers import utils
 from timesketch.lib.datastores.opensearch import OpenSearchDataStore
 
 logger = logging.getLogger("timesketch.data_finder")
 
 
+def _is_valid_iso_date(date_string):
+    """Check whether a string is a valid ISO 8601-formatted date.
+
+    Args:
+        date_string (str): The string to validate.
+
+    Returns:
+        bool: True if the string is a valid ISO 8601 date, False otherwise.
+    """
+    try:
+        datetime.fromisoformat(date_string)
+        return True
+    except (ValueError, TypeError):
+        return False
+
+
 class DataFinder:
     """The data finder class."""
 
@@ -69,8 +86,13 @@ def can_run(self):
         return True
 
     def set_end_date(self, end_date):
-        """Sets the end date of the time period the data finder uses."""
-        # TODO: Implement a check if this is a valid ISO formatted date.
+        """Sets the end date of the time period the data finder uses.
+
+        Args:
+            end_date (str): An ISO 8601-formatted date string.
+        """
+        if not _is_valid_iso_date(end_date):
+            logger.warning("end_date [%s] is not a valid ISO 8601 date.", end_date)
         self._end_date = end_date
 
     def set_indices(self, indices):
@@ -106,8 +128,13 @@ def set_rule(self, rule_dict):
         self._rule = rule_dict
 
     def set_start_date(self, start_date):
-        """Sets the start date of the time period the data finder uses."""
-        # TODO: Implement a check if this is a valid ISO formatted date.
+        """Sets the start date of the time period the data finder uses.
+
+        Args:
+            start_date (str): An ISO 8601-formatted date string.
+        """
+        if not _is_valid_iso_date(start_date):
+            logger.warning("start_date [%s] is not a valid ISO 8601 date.", start_date)
         self._start_date = start_date
 
     def set_timeline_ids(self, timeline_ids):