From 47e156c5e439e1c0b1a1e7aede81ce1dbcc26029 Mon Sep 17 00:00:00 2001 From: Mahmoud Ashraf <182176867+SNO7E-G@users.noreply.github.com> Date: Wed, 25 Mar 2026 21:08:14 +0500 Subject: [PATCH] test: add missing analyzer unit tests and datafinder date validation --- timesketch/lib/analyzers/domain_test.py | 101 ++++++++++++++++++ .../lib/analyzers/feature_extraction_test.py | 27 ++++- .../lib/analyzers/ntfs_timestomp_test.py | 86 ++++++++++++++- timesketch/lib/datafinder.py | 35 +++++- 4 files changed, 239 insertions(+), 10 deletions(-) diff --git a/timesketch/lib/analyzers/domain_test.py b/timesketch/lib/analyzers/domain_test.py index f4b0806458..98838732d2 100644 --- a/timesketch/lib/analyzers/domain_test.py +++ b/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,91 @@ 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) + analyzer.datastore.client = mock.Mock() + + # No events in the store: run() should return early with a NOTE summary. + 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) + analyzer.datastore.client = mock.Mock() + + # Inject a single event that carries a url but no domain field. + event_id = "url_event_0" + analyzer.datastore.event_store[event_id] = { + "_id": event_id, + "_index": "test", + "_source": { + "__ts_timeline_id": 1, + "timestamp": 1410593222543942, + "url": "https://www.example.com/some/path?q=1", + }, + } + + result = analyzer.run() + # example.com should have been recognised and counted as a domain. + 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) + 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). + for i in range(10): + eid = "common_{:d}".format(i) + analyzer.datastore.event_store[eid] = { + "_id": eid, + "_index": "test", + "_source": { + "__ts_timeline_id": 1, + "timestamp": 1410593222543942 + i, + "domain": "common.example.com", + }, + } + + rare_eid = "rare_0" + analyzer.datastore.event_store[rare_eid] = { + "_id": rare_eid, + "_index": "test", + "_source": { + "__ts_timeline_id": 1, + "timestamp": 1410593222543942, + "domain": "rare.example.net", + }, + } + + result = analyzer.run() + # Both the common and the rare domain should be present in the summary. + 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) + analyzer.datastore.client = mock.Mock() + + cdn_eid = "cdn_0" + analyzer.datastore.event_store[cdn_eid] = { + "_id": cdn_eid, + "_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) diff --git a/timesketch/lib/analyzers/feature_extraction_test.py b/timesketch/lib/analyzers/feature_extraction_test.py index 067684d7b1..5d86639c95 100644 --- a/timesketch/lib/analyzers/feature_extraction_test.py +++ b/timesketch/lib/analyzers/feature_extraction_test.py @@ -197,7 +197,32 @@ def _config_validation(self, config): if aggregate: self.assertIsInstance(aggregate, bool) - # TODO: Add tests for the feature extraction. + 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"}) + + 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") + + 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") diff --git a/timesketch/lib/analyzers/ntfs_timestomp_test.py b/timesketch/lib/analyzers/ntfs_timestomp_test.py index 07530409b2..7011c18b83 100644 --- a/timesketch/lib/analyzers/ntfs_timestomp_test.py +++ b/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)) diff --git a/timesketch/lib/datafinder.py b/timesketch/lib/datafinder.py index 7477ff7368..4eae4afb00 100644 --- a/timesketch/lib/datafinder.py +++ b/timesketch/lib/datafinder.py @@ -14,6 +14,7 @@ """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 @@ -21,6 +22,22 @@ 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):