diff --git a/api_client/python/timesketch_api_client/sketch.py b/api_client/python/timesketch_api_client/sketch.py index ad74426d2a..97776ecf24 100644 --- a/api_client/python/timesketch_api_client/sketch.py +++ b/api_client/python/timesketch_api_client/sketch.py @@ -502,13 +502,17 @@ def delete(self, force_delete=False): # Check the return status. If it's not a success (20x), # error_message will raise a RuntimeError. if not error.check_return_status(response, logger): + if response.status_code == definitions.HTTP_STATUS_CODE_NOT_FOUND: + error.error_message( + response, + message=f"Failed to delete sketch {self.id}", + error=error.NotFoundError, + ) error.error_message( response, message=f"Failed to delete sketch {self.id}", - error=RuntimeError, ) - else: - return error.check_return_status(response, logger) + return True def add_to_acl( @@ -991,12 +995,16 @@ def list_timelines(self): return timelines for timeline_dict in objects[0].get("timelines", []): + searchindex = timeline_dict.get("searchindex") + searchindex_name = "" + if isinstance(searchindex, dict): + searchindex_name = searchindex.get("index_name") or "" timeline_obj = timeline.Timeline( - timeline_id=timeline_dict["id"], + timeline_id=timeline_dict.get("id"), sketch_id=self.id, api=self.api, - name=timeline_dict["name"], - searchindex=timeline_dict["searchindex"]["index_name"], + name=timeline_dict.get("name"), + searchindex=searchindex_name, ) timelines.append(timeline_obj) return timelines diff --git a/cli_client/python/timesketch_cli_client/commands/sketch.py b/cli_client/python/timesketch_cli_client/commands/sketch.py index 4ded022bc4..97e6e08f24 100644 --- a/cli_client/python/timesketch_cli_client/commands/sketch.py +++ b/cli_client/python/timesketch_cli_client/commands/sketch.py @@ -23,6 +23,7 @@ from timesketch_cli_client.commands import attribute as attribute_command from timesketch_api_client import search +from timesketch_api_client.error import NotFoundError @click.group("sketch") @@ -328,31 +329,70 @@ def delete_sketch(ctx: click.Context, force_delete: bool) -> None: force_delete: If true, delete immediately. """ sketch = ctx.obj.sketch - # if sketch is archived, exit - if sketch.is_archived(): - click.echo("Error Sketch is archived") + + # Initialize with default values. Preserve cached sketch_name if it exists. + sketch_name = getattr(sketch, "_sketch_name", None) or "" + sketch_desc = "N/A" + sketch_status = "N/A" + sketch_labels = "N/A" + timelines = [] + + try: + is_archived = sketch.is_archived() + except NotFoundError: + click.echo( + f"Error: Sketch {sketch.id} not found or you do not have permission " + f"to access it." + ) + ctx.exit(1) + + if is_archived: + click.echo("Error: Sketch is archived.") ctx.exit(1) + try: + sketch_name = sketch.name + sketch_desc = sketch.description + sketch_status = sketch.status + sketch_labels = sketch.labels + timelines = sketch.list_timelines() + except NotFoundError: + pass + # Dryrun: if not force_delete: click.echo("Would delete the following things (use --force_delete to execute)") + click.echo( - f"Sketch: {sketch.id} {sketch.name} {sketch.description} {sketch.status} Labels: {sketch.labels}" # pylint: disable=line-too-long + f"Sketch: {sketch.id} {sketch_name} {sketch_desc} {sketch_status} Labels: {sketch_labels}" # pylint: disable=line-too-long ) - for timeline in sketch.list_timelines(): + for timeline in timelines: + timeline_desc = "N/A" + timeline_status = "N/A" + try: + # timeline.description and timeline.status lazy-load from the API. + timeline_desc = timeline.description + timeline_status = timeline.status + except NotFoundError: + pass click.echo( - f" Timeline: {timeline.id} {timeline.name} {timeline.description} {timeline.status}" # pylint: disable=line-too-long + f" Timeline: {timeline.id} {timeline.name} {timeline_desc} {timeline_status}" # pylint: disable=line-too-long ) if force_delete: # --- Check the response for success or error --- try: sketch.delete(force_delete=force_delete) - click.echo(f"Sketch {sketch.id} '{sketch.name}' successfully deleted.") + click.echo(f"Sketch {sketch.id} '{sketch_name}' successfully deleted.") + except NotFoundError: + click.echo( + f"Failed to delete sketch {sketch.id} '{sketch_name}'. Error: Sketch was not found (perhaps already permanently deleted?)." # pylint: disable=line-too-long + ) + ctx.exit(1) except RuntimeError as e: click.echo( - f"Failed to delete sketch {sketch.id} '{sketch.name}'. Error: {e}" + f"Failed to delete sketch {sketch.id} '{sketch_name}'. Error: {e}" ) ctx.exit(1) diff --git a/end_to_end_tests/cli_client_e2e_test.py b/end_to_end_tests/cli_client_e2e_test.py index 715c591873..dac45b0471 100644 --- a/end_to_end_tests/cli_client_e2e_test.py +++ b/end_to_end_tests/cli_client_e2e_test.py @@ -215,6 +215,49 @@ def test_cli_integration(self): self.assertions.assertEqual(result.exit_code, 0, f"Failed: {result.output}") self.assertions.assertEqual(result.output.strip(), "42") + def test_cli_sketch_delete_soft_deleted(self): + """Tests that a soft-deleted sketch can be deleted via CLI.""" + # Create a new sketch to be soft-deleted and then hard-deleted + sketch_name = f"cli_soft_delete_test_{uuid.uuid4().hex}" + sketch = self.api.create_sketch(name=sketch_name) + + # We need a timeline to trigger the loop in sketch.py + self.import_timeline("evtx_part.csv", sketch=sketch) + + # Soft delete the sketch + sketch.delete(force_delete=False) + + # Get a fresh instance of the sketch so cached values are cleared + fresh_sketch = self.api.get_sketch(sketch.id) + + # Now try to delete it via CLI (dry-run first, then force) + cli_ctx_obj = E2ECliContextObject( + api_client=self.api, + sketch_instance=fresh_sketch, + output_format="text", + ) + + # Dry-run + result = self.runner.invoke(sketch_group, ["delete"], obj=cli_ctx_obj) + self.assertions.assertEqual( + result.exit_code, + 1, + f"CLI command 'sketch delete' (dry-run) failed to exit with 1 on soft-deleted sketch.\nOutput:\n{result.output}\nException:\n{result.exception}", # pylint: disable=line-too-long + ) + + # Force-delete + result_force = self.runner.invoke( + sketch_group, ["delete", "--force_delete"], obj=cli_ctx_obj + ) + self.assertions.assertEqual( + result_force.exit_code, + 1, + f"CLI command 'sketch delete --force_delete' unexpectedly succeeded on a soft-deleted sketch for a non-admin.\nOutput:\n{result_force.output}\nException:\n{result_force.exception}", # pylint: disable=line-too-long + ) + self.assertions.assertIn( + "not found or you do not have permission", result_force.output + ) + # Register the new test class with the test manager manager.EndToEndTestManager.register_test(CliClientE2ETest) diff --git a/timesketch/api/v1/resources/sketch.py b/timesketch/api/v1/resources/sketch.py index 388d99d01a..85f0d799d0 100644 --- a/timesketch/api/v1/resources/sketch.py +++ b/timesketch/api/v1/resources/sketch.py @@ -328,6 +328,12 @@ def _get_sketch_for_admin(sketch: Sketch): A limited view of a sketch in JSON (instance of flask.wrappers.Response) """ + if not current_user.admin: + abort( + HTTP_STATUS_CODE_FORBIDDEN, + "Admin privileges required to access this view.", + ) + if sketch.get_status.status == "archived": status = "archived" elif sketch.get_status.status == "deleted": @@ -335,12 +341,53 @@ def _get_sketch_for_admin(sketch: Sketch): else: status = "admin_view" + has_read_permission = sketch.has_permission(current_user, "read") + timelines = [] + for timeline in sketch.timelines: + timeline_status = ( + timeline.get_status.status if timeline.get_status else "unknown" + ) + if has_read_permission: + timelines.append( + { + "id": timeline.id, + "name": timeline.name, + "description": timeline.description, + "color": timeline.color, + "status": [{"id": 0, "status": timeline_status}], + "searchindex": { + "index_name": ( + timeline.searchindex.index_name + if timeline.searchindex + else "" + ) + }, + "created_at": timeline.created_at, + "updated_at": timeline.updated_at, + "deleted": timeline_status == "deleted", + } + ) + else: # for admins, only redacted information are needed + timelines.append( + { + "id": timeline.id, + "name": "", + "description": "", + "color": "", + "status": [{"id": 0, "status": timeline_status}], + "searchindex": {"index_name": ""}, + "created_at": timeline.created_at, + "updated_at": timeline.updated_at, + "deleted": timeline_status == "deleted", + } + ) + sketch_fields = { "id": sketch.id, "name": sketch.name, "description": sketch.description, "user": {"username": current_user.username}, - "timelines": [], + "timelines": timelines, "stories": [], "active_timelines": [], "label_string": sketch.label_string, diff --git a/timesketch/api/v1/resources_test.py b/timesketch/api/v1/resources_test.py index cd430bc3f7..1da48ae528 100644 --- a/timesketch/api/v1/resources_test.py +++ b/timesketch/api/v1/resources_test.py @@ -18,6 +18,7 @@ import tempfile import json from unittest import mock +from werkzeug.exceptions import Forbidden from timesketch.lib.definitions import HTTP_STATUS_CODE_BAD_REQUEST from timesketch.lib.definitions import HTTP_STATUS_CODE_CREATED @@ -46,6 +47,7 @@ from timesketch.lib.llms.providers import manager as llm_manager from timesketch.lib.llms.features import interface as feature_interface from timesketch.api.v1.resources import llm +from timesketch.api.v1.resources import sketch as sketch_resources class ResourceMixinTest(BaseTest): @@ -271,6 +273,46 @@ def test_sketch_acl(self): response = self.client.get("/api/v1/sketches/2/") self.assert403(response) + def test_get_sketch_for_admin_no_read_permission(self): + """Admin request to get a sketch without explicit read permission.""" + self.login_admin() + response = self.client.get("/api/v1/sketches/1/") + self.assert200(response) + sketch_obj = response.json["objects"][0] + self.assertEqual(sketch_obj["id"], 1) + self.assertEqual(sketch_obj["status"][0]["status"], "admin_view") + self.assertEqual(len(sketch_obj["timelines"]), 1) + self.assertEqual(sketch_obj["timelines"][0]["name"], "") + self.assertEqual(sketch_obj["timelines"][0]["description"], "") + + @mock.patch("timesketch.api.v1.resources.OpenSearchDataStore", MockDataStore) + def test_get_sketch_for_admin_soft_deleted(self): + """Admin request to get a soft-deleted sketch with read permission.""" + # Grant admin explicit read permission on sketch 1 + self.sketch1.grant_permission(permission="read", user=self.useradmin) + self.login_admin() + + # Soft-delete sketch 1 + delete_response = self.client.delete(self.resource_url) + self.assert200(delete_response) + + # Admin gets the soft-deleted sketch + response = self.client.get(self.resource_url) + self.assert200(response) + sketch_obj = response.json["objects"][0] + self.assertEqual(sketch_obj["status"][0]["status"], "deleted") + self.assertEqual(len(sketch_obj["timelines"]), 1) + self.assertEqual(sketch_obj["timelines"][0]["name"], "Timeline 1") + + def test_get_sketch_for_admin_non_admin_forbidden(self): + """Non-admin invocation of _get_sketch_for_admin is rejected.""" + self.login() + with self.client: + self.client.get("/") + # pylint: disable=protected-access + with self.assertRaises(Forbidden): + sketch_resources.SketchResource._get_sketch_for_admin(self.sketch1) + def test_create_a_sketch(self): """Authenticated request to create a sketch.""" self.login()