diff --git a/api_client/python/timesketch_api_client/client.py b/api_client/python/timesketch_api_client/client.py index 621675e0fd..03d2ec1212 100644 --- a/api_client/python/timesketch_api_client/client.py +++ b/api_client/python/timesketch_api_client/client.py @@ -688,6 +688,47 @@ def get_sketch(self, sketch_id): """ return sketch.Sketch(sketch_id, api=self) + def get_sketches_by_name( + self, sketch_name: str, include_archived: bool = True + ) -> list[sketch.Sketch]: + """Get all sketches with an exact matching name. + + The lookup runs server side with the "search" scope, i.e. it covers + all sketches the user can read (owned, shared and public). The server + search is a partial, case-insensitive match on name and description, + so the results are filtered here for an exact, case-sensitive name + match. + + Args: + sketch_name (str): The name of the sketch to find. + Warning: Timesketch allows multiple sketches with the same name. + The caller is responsible for handling this potential conflict. + include_archived (bool): If archived sketches should be returned. + Defaults to True. + + Raises: + KeyError: If no sketch with the specified name is found. + + Returns: + list[sketch.Sketch]: A list of sketch objects. + """ + # We still need to verify the match for the sketch name, as the search_query + # also matches on partial (ILIKE) matches by default + sketches = [ + sketch_obj + for sketch_obj in self.list_sketches( + search_query=sketch_name, + scope="search", + include_archived=include_archived, + ) + if sketch_obj.name == sketch_name + ] + + if not sketches: + raise KeyError(f"Sketch with name '{sketch_name}' not found.") + + return sketches + def get_aggregator_info(self, name="", as_pandas=False): """Returns information about available aggregators. @@ -734,12 +775,19 @@ def get_aggregator_info(self, name="", as_pandas=False): return pandas.DataFrame(lines) - def list_sketches(self, per_page=50, scope="user", include_archived=True): + def list_sketches( + self, + per_page=50, + scope=None, + include_archived=True, + search_query=None, + ): """Get a list of all open sketches that the user has access to. Args: per_page (int): Number of items per page when paginating. Default is 50. - scope (str): What scope to get sketches as. Default to user. + scope (str): What scope to get sketches as. Defaults to "search" if + search_query is specified, otherwise to "user". user: sketches owned by the user recent: sketches that the user has actively searched in shared: sketches shared with the user (but not owned by them) @@ -748,15 +796,33 @@ def list_sketches(self, per_page=50, scope="user", include_archived=True): search: pass additional search query all: all sketches the user has access to (owned and shared) include_archived (bool): If archived sketches should be returned. + search_query (str): A query string to search for sketches by name + or description. Only supported with the "search" scope. The + server performs a partial, case-insensitive match, so results + may include sketches whose name only contains the query. + + Raises: + ValueError: If search_query is used with a scope other than "search". Yields: Sketch objects instances. """ + if scope is None: + scope = "search" if search_query else "user" + + if search_query and scope != "search": + raise ValueError( + f"search_query is only supported with scope 'search', not '{scope}'." + ) + url_params = { "per_page": per_page, "scope": scope, "include_archived": include_archived, } + + if search_query: + url_params["search_query"] = search_query # Start with the first page page = 1 has_next_page = True diff --git a/api_client/python/timesketch_api_client/client_test.py b/api_client/python/timesketch_api_client/client_test.py index afcdadd43f..7a485678ad 100644 --- a/api_client/python/timesketch_api_client/client_test.py +++ b/api_client/python/timesketch_api_client/client_test.py @@ -77,6 +77,79 @@ def test_get_sketches(self): self.assertEqual(len(sketches), 1) self.assertIsInstance(sketches[0], sketch_lib.Sketch) + def test_get_sketches_by_name(self): + """Test to get sketches by name.""" + sketches = self.api_client.get_sketches_by_name("test") + self.assertIsInstance(sketches, list) + self.assertEqual(len(sketches), 1) + self.assertIsInstance(sketches[0], sketch_lib.Sketch) + self.assertEqual(sketches[0].name, "test") + + def test_get_sketches_by_name_not_found(self): + """Test that get_sketches_by_name raises KeyError for unknown name.""" + with self.assertRaises(KeyError): + self.api_client.get_sketches_by_name("nonexistent_sketch") + + @mock.patch("requests.Session", test_lib.mock_session) + def test_get_sketches_by_name_duplicates(self): + """Test get_sketches_by_name returns multiple sketches with same name.""" + with mock.patch.object(self.api_client, "list_sketches") as mock_list: + sketch_1 = sketch_lib.Sketch( + sketch_id=1, api=self.api_client, sketch_name="duplicate" + ) + sketch_2 = sketch_lib.Sketch( + sketch_id=2, api=self.api_client, sketch_name="duplicate" + ) + mock_list.return_value = iter([sketch_1, sketch_2]) + + sketches = self.api_client.get_sketches_by_name("duplicate") + self.assertIsInstance(sketches, list) + self.assertEqual(len(sketches), 2) + self.assertEqual(sketches[0].id, 1) + self.assertEqual(sketches[1].id, 2) + + @mock.patch("requests.Session", test_lib.mock_session) + def test_get_sketches_by_name_case_sensitive(self): + """Test that get_sketches_by_name matching is case-sensitive.""" + with mock.patch.object(self.api_client, "list_sketches") as mock_list: + sketch_1 = sketch_lib.Sketch( + sketch_id=1, api=self.api_client, sketch_name="My Sketch" + ) + mock_list.return_value = iter([sketch_1]) + + with self.assertRaises(KeyError): + self.api_client.get_sketches_by_name("my sketch") + + def test_get_sketches_by_name_include_archived(self): + """Test that get_sketches_by_name passes include_archived through.""" + with mock.patch.object(self.api_client, "list_sketches") as mock_list: + mock_list.return_value = iter( + [sketch_lib.Sketch(sketch_id=1, api=self.api_client, sketch_name="a")] + ) + self.api_client.get_sketches_by_name("a", include_archived=False) + mock_list.assert_called_once_with( + search_query="a", scope="search", include_archived=False + ) + + def test_list_sketches_scope_defaults(self): + """Test the scope that list_sketches sends to the server by default.""" + with mock.patch.object( + self.api_client, "fetch_resource_data", return_value={} + ) as mock_fetch: + list(self.api_client.list_sketches()) + self.assertEqual(mock_fetch.call_args.kwargs["params"]["scope"], "user") + self.assertNotIn("search_query", mock_fetch.call_args.kwargs["params"]) + + list(self.api_client.list_sketches(search_query="foo")) + params = mock_fetch.call_args.kwargs["params"] + self.assertEqual(params["scope"], "search") + self.assertEqual(params["search_query"], "foo") + + def test_list_sketches_search_query_with_wrong_scope(self): + """Test that search_query cannot be combined with a non-search scope.""" + with self.assertRaises(ValueError): + list(self.api_client.list_sketches(scope="user", search_query="foo")) + class TimesketchApiRetryTest(unittest.TestCase): """Test TimesketchApi client retry logic.""" diff --git a/api_client/python/timesketch_api_client/sketch.py b/api_client/python/timesketch_api_client/sketch.py index 3e6b3ac5b8..6de14db68d 100644 --- a/api_client/python/timesketch_api_client/sketch.py +++ b/api_client/python/timesketch_api_client/sketch.py @@ -161,6 +161,42 @@ def last_activity(self): meta = data.get("meta", {}) return meta.get("last_activity", "") + @property + def created_at(self): + """Property that returns sketch creation time. + + Returns: + str: Sketch creation time as string. + """ + data = self.lazyload_data() + objects = data.get("objects") + if not objects: + return "" + data_object = objects[0] + created_at_string = data_object.get("created_at", "") + return created_at_string + + @property + def creator(self): + """Property that returns sketch creator. + + Returns: + str: Sketch creator as string. + """ + data = self.lazyload_data() + objects = data.get("objects") + if not objects: + return "" + data_object = objects[0] + try: + username_string = data_object.get("user").get("username") + if not username_string: + return "" + return username_string + except (AttributeError, TypeError) as e: + logger.error("Error getting sketch creator: %s", e) + return "" + @property def my_acl(self): """Property that returns back the ACL for the current user.""" diff --git a/api_client/python/timesketch_api_client/sketch_test.py b/api_client/python/timesketch_api_client/sketch_test.py index 7921cd2779..6f13d264ed 100644 --- a/api_client/python/timesketch_api_client/sketch_test.py +++ b/api_client/python/timesketch_api_client/sketch_test.py @@ -79,6 +79,16 @@ def test_add_event_attributes(self): response = self.sketch.add_event_attributes(events) self.assertEqual(response, expected_response) + def test_created_at(self): + """Test the created_at property.""" + created_at = self.sketch.created_at + self.assertEqual(created_at, "2025-12-11T07:39:20.000000") + + def test_creator(self): + """Test the creator property.""" + creator = self.sketch.creator + self.assertEqual(creator, "testuser") + def test_add_event_attributes_invalid(self): """Confirm an exception is raised when events isn't a list.""" events = {"_id": "1", "_type": "_doc", "index": "1", "attributes": []} diff --git a/api_client/python/timesketch_api_client/test_lib.py b/api_client/python/timesketch_api_client/test_lib.py index 0912d21cb1..246fda6d45 100644 --- a/api_client/python/timesketch_api_client/test_lib.py +++ b/api_client/python/timesketch_api_client/test_lib.py @@ -192,6 +192,8 @@ def json(self): "id": 1, "name": "test", "description": "test", + "created_at": "2025-12-11T07:39:20.000000", + "user": {"username": "testuser"}, "timelines": [ { "id": 1, diff --git a/docs/guides/user/upload-data.md b/docs/guides/user/upload-data.md index 2b1443277b..57ae11320e 100644 --- a/docs/guides/user/upload-data.md +++ b/docs/guides/user/upload-data.md @@ -61,6 +61,23 @@ Other parameters suggested to be set are `sketch_id` (if it isn't provided a new sketch will be created) and `timeline_name` (otherwise a default name will be chosen). +Alternatively, you can use `--sketch-name` to target a sketch by name. If a +sketch with that name already exists, the importer will reuse it instead of +creating a new one. Since sketch names are not unique in Timesketch, you can +control how duplicates are handled with the `--sketch-strategy` flag: + +- `ask` (default): prompts the user to select the correct sketch interactively. + If no terminal is available (e.g. in scripts or pipelines), the importer + exits with an error listing the matching sketch IDs instead. +- `newest`: automatically picks the most recently created sketch (highest ID). +- `oldest`: automatically picks the earliest created sketch (lowest ID). + +Example: + +```shell +$ timesketch_importer.py --sketch-name "My Investigation" --sketch-strategy newest path_to_my_file.csv +``` + Additionally, for Plaso files, you can filter events during import using the `--plaso-event-filter` option, providing a standard Plaso filter string (e.g., `'data_type is "fs:stat"'`). diff --git a/importer_client/python/setup.py b/importer_client/python/setup.py index 12f589f87f..2619679ae3 100644 --- a/importer_client/python/setup.py +++ b/importer_client/python/setup.py @@ -51,6 +51,8 @@ [ "pandas", "xlrd", + # TODO: Pin to "timesketch-api-client>20260611" with the next API + # client release. The importer needs get_sketches_by_name(). "timesketch-api-client", "pyyaml", ] diff --git a/importer_client/python/tools/timesketch_importer.py b/importer_client/python/tools/timesketch_importer.py index 82981b51dd..27826c05d5 100644 --- a/importer_client/python/tools/timesketch_importer.py +++ b/importer_client/python/tools/timesketch_importer.py @@ -352,8 +352,28 @@ def main(args=None): dest="sketch_name", default="", help=( - "String that will be used as the sketch name in case a new " - "sketch is created." + "Name of the sketch to import into. If a non-archived sketch " + "with this exact name exists, it will be reused, otherwise a " + "new sketch with this name is created. If multiple sketches " + "share the name, see --sketch_strategy. If neither --sketch_id " + "nor --sketch_name is provided, a new sketch is always created." + ), + ) + + config_group.add_argument( + "--sketch_strategy", + "--sketch-strategy", + action="store", + type=str, + choices=["ask", "newest", "oldest"], + dest="sketch_strategy", + default="ask", + help=( + "Strategy to use when a sketch name is provided and a sketch " + "with the same name already exists. Supported strategies are: " + "'ask' (default) which will ask the user for input, 'newest' " + "which will use the most recently created sketch, and 'oldest' " + "which will use the earliest created sketch." ), ) @@ -632,19 +652,108 @@ def main(args=None): ) sketch_id = options.sketch_id + sketch_name = options.sketch_name + my_sketch = None if sketch_id: my_sketch = ts_client.get_sketch(sketch_id) - else: - sketch_name = options.sketch_name or "New Sketch From Importer CLI" - my_sketch = ts_client.create_sketch(sketch_name) + elif sketch_name: + # Only look up existing sketches when a name was explicitly provided. + # Archived sketches are excluded since data cannot be uploaded to them. + try: + sketches = ts_client.get_sketches_by_name( + sketch_name, include_archived=False + ) + except KeyError: + sketches = [] + + if len(sketches) > 1: + # Sketch IDs are auto-incremented on creation, so the highest ID is + # the newest sketch. Using the ID avoids a full sketch API request + # per duplicate that would be needed to read created_at. + if options.sketch_strategy == "newest": + my_sketch = max(sketches, key=lambda s: s.id) + elif options.sketch_strategy == "oldest": + my_sketch = min(sketches, key=lambda s: s.id) + elif not sys.stdin.isatty(): + # Never silently pick a sketch. Without a terminal we cannot + # prompt, so fail and tell the user how to resolve it. + logger.error( + "Multiple sketches found with the name '%s' (IDs: %s) and " + "no terminal available to ask which one to use. Use " + "--sketch_id to select one, or --sketch_strategy " + "newest|oldest to resolve duplicates automatically.", + sketch_name, + ", ".join(str(s.id) for s in sorted(sketches, key=lambda s: s.id)), + ) + sys.exit(1) + else: + # ask user for clarification using cli_input + print(f"Multiple sketches found with the name '{sketch_name}':") + for s in sketches: + print(f" - [{s.id:d}] created_at: {s.created_at}, by {s.creator}") + + valid_ids = {s.id for s in sketches} + selected_option = None + while selected_option is None: + try: + ans = cli_input.ask_question( + "Select the sketch to use by entering " + "the corresponding number", + input_type=int, + default=sketches[0].id, + ) + if ans in valid_ids: + selected_option = ans + else: + print( + "Invalid ID. Please choose from the listed IDs: " + f"{sorted(valid_ids)}" + ) + except ValueError: + print( + "Invalid input. Please enter a valid integer " + f"(e.g. {sketches[0].id:d})." + ) + my_sketch = next(s for s in sketches if s.id == selected_option) + elif sketches: + my_sketch = sketches[0] + + if my_sketch: + logger.info( + "Using existing sketch: [%d] %s", + my_sketch.id, + my_sketch.name, + ) + + if not my_sketch and not sketch_id: + # Either no name was given, or no sketch with that name exists. Never + # reuse a sketch based on the default name, always create a new one. + my_sketch = ts_client.create_sketch( + sketch_name or "New Sketch From Importer CLI" + ) logger.info( - "New sketch created: [{0:d}] {1:s}".format(my_sketch.id, my_sketch.name) + "New sketch created: [%d] %s", + my_sketch.id, + my_sketch.name, ) if not my_sketch: logger.error("Unable to get sketch ID: {0:d}".format(sketch_id)) sys.exit(1) + # Fail early with a clear message instead of a failed upload (HTTP 403) + # if the user is not allowed to add data to the selected sketch. + sketch_acl = my_sketch.my_acl + if sketch_acl and "write" not in sketch_acl: + logger.error( + "You do not have write access to sketch [%d] %s, unable to import " + "data. Ask the sketch owner for write access or choose a different " + "sketch with --sketch_id or --sketch_name.", + my_sketch.id, + my_sketch.name, + ) + sys.exit(1) + filename = os.path.basename(options.path) default_timeline_name, _, _ = filename.rpartition(".")