Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 14 additions & 6 deletions api_client/python/timesketch_api_client/sketch.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down Expand Up @@ -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
Expand Down
56 changes: 48 additions & 8 deletions cli_client/python/timesketch_cli_client/commands/sketch.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down Expand Up @@ -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 "<Unknown/Deleted>"
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()

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If I'm not mistaken, the API (_get_sketch_for_admin) returns timelines: []. This means for admins, the dry-run here will show no timelines, but --force_delete will still permanently delete them from the DB.

That seems a bit risky because the admin won't see what they are actually deleting. Or is this intentional?

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

oh yeah it might be good to at least mention the id and status of the timeline, we do not need to expose the name and description for admins.

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

DOne ptal, I now return redacted info and I also added another permission check in the get_sketch_for_admin just to be on the safe side

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:
Comment thread
jaegeral marked this conversation as resolved.
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)

Expand Down
43 changes: 43 additions & 0 deletions end_to_end_tests/cli_client_e2e_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
49 changes: 48 additions & 1 deletion timesketch/api/v1/resources/sketch.py
Original file line number Diff line number Diff line change
Expand Up @@ -328,19 +328,66 @@ 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":
status = "deleted"
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": "<Restricted>",
"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,
Expand Down
42 changes: 42 additions & 0 deletions timesketch/api/v1/resources_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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):
Expand Down Expand Up @@ -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"], "<Restricted>")
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()
Expand Down
Loading