🐾 fix(api): workspace DELETE must not report success it didn't perform

DELETE returned 204 for a library owned by another user, having deleted
nothing. The caller (Daedalus) recorded the workspace as cleaned up and
dropped its own row, while the Library survived holding its globally-unique
name — so that name could never be reused, and nothing anywhere recorded why.

An unowned library now returns 409 owner_conflict, reusing the code the create
path already emits for the same condition. Ownership stays opaque on GET (404
as before): the disclosure concern is about reads, and a delete that silently
does nothing is the worse failure.

A genuinely absent library still returns 204 — that idempotency is relied upon
and is correct.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-08-07 13:03:32 -04:00
parent b5fba26d4c
commit e10331dd44
2 changed files with 108 additions and 6 deletions

View File

@@ -213,23 +213,40 @@ def workspace_detail_or_delete(request, workspace_id):
except Library.DoesNotExist: except Library.DoesNotExist:
lib = None lib = None
# Cross-user reads/writes look like "not found" — don't disclose # Cross-user reads look like "not found" — don't disclose existence
# existence across users. # across users. DELETE is handled separately below: telling the caller
if lib is not None and lib.owner_username != request.user.username: # "deleted" about a library we did not touch is what silently orphans
lib = None # libraries, and an orphan holds its globally-unique name forever.
unowned = lib is not None and lib.owner_username != request.user.username
if request.method == "GET": if request.method == "GET":
if lib is None: if lib is None or unowned:
return Response( return Response(
{"detail": "Workspace not found."}, {"detail": "Workspace not found."},
status=status.HTTP_404_NOT_FOUND, status=status.HTTP_404_NOT_FOUND,
) )
return Response(WorkspaceStatusSerializer(_serialize_workspace(lib)).data) return Response(WorkspaceStatusSerializer(_serialize_workspace(lib)).data)
# DELETE — idempotent: a missing (or unowned) workspace returns 204. # DELETE — idempotent only where nothing exists to delete.
if lib is None: if lib is None:
return Response(status=status.HTTP_204_NO_CONTENT) return Response(status=status.HTTP_204_NO_CONTENT)
if unowned:
# Still opaque about ownership, but never a false success: the
# caller must not record this workspace as cleaned up.
logger.warning(
"workspace_delete owner_conflict workspace_id=%s library_uid=%s "
"caller=%s",
workspace_id, lib.uid, request.user.username,
)
return Response(
{
"detail": "Workspace id is already in use.",
"code": "owner_conflict",
},
status=status.HTTP_409_CONFLICT,
)
# Delete the Library and everything reachable + unique to it, plus # Delete the Library and everything reachable + unique to it, plus
# orphan-Concept GC. Shared with the admin/HTML delete path. # orphan-Concept GC. Shared with the admin/HTML delete path.
result = delete_library_cascade(lib) result = delete_library_cascade(lib)

View File

@@ -7,6 +7,9 @@ search scoping) require Neo4j and are validated by the manual end-to-end
test plan, not these unit tests. test plan, not these unit tests.
""" """
from unittest.mock import Mock, patch
from django.contrib.auth.models import User
from django.test import TestCase from django.test import TestCase
from rest_framework.test import APIClient from rest_framework.test import APIClient
@@ -208,3 +211,85 @@ class WorkspaceEndpointAuthTests(TestCase):
def test_workspace_delete_requires_auth(self): def test_workspace_delete_requires_auth(self):
response = self.client.delete("/library/api/workspaces/ws_a/") response = self.client.delete("/library/api/workspaces/ws_a/")
self.assertIn(response.status_code, [401, 403]) self.assertIn(response.status_code, [401, 403])
class WorkspaceDeleteOwnershipTests(TestCase):
"""DELETE must never report success for a library it did not delete.
A false 204 is how libraries get orphaned: Daedalus records the
workspace as cleaned up and drops its own row, while the Library node
survives holding its globally-unique name forever — which then blocks
ever recreating a workspace under that name.
"""
def setUp(self):
self.client = APIClient()
self.user = User.objects.create_user(
username="owner", password="pw" # noqa: S106 — test credential
)
self.client.force_authenticate(user=self.user)
def _library(self, owner_username):
lib = Mock()
lib.uid = "lib_1"
lib.owner_username = owner_username
return lib
def test_absent_library_still_returns_204(self):
"""Genuine idempotency is preserved — nothing exists, nothing to do."""
from library.models import Library
with patch(
"neomodel.sync_.match.NodeSet.get",
side_effect=Library.DoesNotExist("nope"),
):
response = self.client.delete("/library/api/workspaces/ws_gone/")
self.assertEqual(response.status_code, 204)
def test_unowned_library_returns_409_and_is_not_deleted(self):
from library.models import Library
lib = self._library("someone_else")
with (
patch("neomodel.sync_.match.NodeSet.get", return_value=lib),
patch(
"library.api.workspaces.delete_library_cascade"
) as cascade,
):
response = self.client.delete("/library/api/workspaces/ws_a/")
self.assertEqual(response.status_code, 409)
self.assertEqual(response.json()["code"], "owner_conflict")
cascade.assert_not_called()
def test_owned_library_is_deleted(self):
from library.models import Library
lib = self._library("owner")
with (
patch("neomodel.sync_.match.NodeSet.get", return_value=lib),
patch(
"library.api.workspaces.delete_library_cascade",
return_value={
"library_uid": "lib_1",
"name": "Assistant",
"item_count": 0,
"orphans_deleted": 0,
},
) as cascade,
):
response = self.client.delete("/library/api/workspaces/ws_a/")
self.assertEqual(response.status_code, 204)
cascade.assert_called_once_with(lib)
def test_unowned_library_get_still_looks_absent(self):
"""Ownership must stay opaque on reads — 404, not 409."""
from library.models import Library
lib = self._library("someone_else")
with patch("neomodel.sync_.match.NodeSet.get", return_value=lib):
response = self.client.get("/library/api/workspaces/ws_a/")
self.assertEqual(response.status_code, 404)