feat(users): admin can permanently delete users, fix set-password redirect bug
Aegis CI / lint-and-test (push) Has been cancelled
Snyk Security Scan / Python vulnerabilities (backend) (push) Has been cancelled
Snyk Security Scan / npm vulnerabilities (frontend) (push) Has been cancelled
Snyk Security Scan / Docker image vulnerabilities (backend) (push) Has been cancelled
Aegis CI / lint-and-test (push) Has been cancelled
Snyk Security Scan / Python vulnerabilities (backend) (push) Has been cancelled
Snyk Security Scan / npm vulnerabilities (frontend) (push) Has been cancelled
Snyk Security Scan / Docker image vulnerabilities (backend) (push) Has been cancelled
- New DELETE /users/{id}: hard-deletes a user only if they have zero
activity footprint (no tests, evidence, worklogs, audit entries, Jira
links, procedure/template suggestions, reviewed imports) — otherwise
rejects with a clear message to deactivate instead, to keep the audit
trail intact. Cleans up the user's own notifications/password-setup
tokens as part of deletion. Self-delete is blocked. Wired to a trash
icon in the Users page with a confirm dialog.
- Registered the EvaluationImport model in app/models/__init__.py — it
was missing, so its table never existed in the SQLite test DB; only
surfaced once the new delete-user footprint check queried it.
- Fixed a redirect bug: the axios response interceptor's on-401 redirect
to /login didn't exempt /set-password, so the auth provider's routine
mount-time /auth/me check (401 for a logged-out visitor) bounced anyone
opening their emailed set-password link away before they could use it.
This commit is contained in:
@@ -49,6 +49,7 @@ from app.models.procedure_suggestion import ProcedureSuggestion
|
||||
from app.models.template_suggestion import TemplateSuggestion
|
||||
from app.models.user import User
|
||||
from app.models.password_setup_token import PasswordSetupToken
|
||||
from app.models.evaluation_import import EvaluationImport
|
||||
|
||||
# Assign __all__ = [
|
||||
__all__ = [
|
||||
@@ -94,4 +95,5 @@ __all__ = [
|
||||
"ProcedureSuggestion",
|
||||
"TemplateSuggestion",
|
||||
"PasswordSetupToken",
|
||||
"EvaluationImport",
|
||||
]
|
||||
|
||||
@@ -36,6 +36,7 @@ from app.services.audit_service import log_action
|
||||
# Import from app.services.user_service
|
||||
from app.services.user_service import (
|
||||
create_user_without_password,
|
||||
delete_user,
|
||||
get_user_or_raise,
|
||||
list_users,
|
||||
switch_active_role,
|
||||
@@ -307,6 +308,36 @@ def update_user_route(
|
||||
return user
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# DELETE /users/{id} — permanently delete a user
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||||
def delete_user_route(
|
||||
user_id: uuid.UUID,
|
||||
db: Session = Depends(get_db),
|
||||
current_user: User = Depends(require_role("admin")),
|
||||
) -> None:
|
||||
"""Permanently delete a user. **Requires admin role.**.
|
||||
|
||||
Only users with no activity footprint (no tests, evidence, worklogs,
|
||||
audit entries, etc.) can be hard-deleted — anyone else must be
|
||||
deactivated instead, to preserve the audit trail.
|
||||
"""
|
||||
with UnitOfWork(db) as uow:
|
||||
delete_user(db, user_id, current_user.id)
|
||||
log_action(
|
||||
db,
|
||||
user_id=current_user.id,
|
||||
action="delete_user",
|
||||
entity_type="user",
|
||||
entity_id=user_id,
|
||||
details={},
|
||||
)
|
||||
uow.commit()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# POST /users/{id}/send-password-email — set-password / reset link
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -180,6 +180,73 @@ def update_user(db: Session, user_id: uuid.UUID, **fields: object) -> User:
|
||||
return user
|
||||
|
||||
|
||||
def delete_user(db: Session, user_id: uuid.UUID, actor_id: uuid.UUID) -> None:
|
||||
"""Permanently delete a user — only if they have no activity footprint.
|
||||
|
||||
Raises EntityNotFoundError if the user doesn't exist. Raises
|
||||
BusinessRuleViolation if the caller is trying to delete themselves, or
|
||||
if the user has any historical footprint (tests, evidence, worklogs,
|
||||
audit entries, Jira links, procedure/template suggestions, reviewed
|
||||
imports) — deleting those users would either violate FK constraints or
|
||||
silently erase audit trail we're required to keep; deactivate them
|
||||
instead. Notifications and pending password-setup tokens are personal-
|
||||
only records and are cleaned up as part of the deletion.
|
||||
|
||||
Does not commit; caller commits.
|
||||
"""
|
||||
user = get_user_or_raise(db, user_id)
|
||||
|
||||
if user_id == actor_id:
|
||||
raise BusinessRuleViolation("You cannot delete your own account.")
|
||||
|
||||
from app.models.audit import AuditLog
|
||||
from app.models.evaluation_import import EvaluationImport
|
||||
from app.models.evidence import Evidence
|
||||
from app.models.jira_link import JiraLink
|
||||
from app.models.notification import Notification
|
||||
from app.models.password_setup_token import PasswordSetupToken
|
||||
from app.models.procedure_suggestion import ProcedureSuggestion
|
||||
from app.models.template_suggestion import TemplateSuggestion
|
||||
from app.models.test import Test
|
||||
from app.models.test_round_history import TestRoundHistory
|
||||
from app.models.worklog import Worklog
|
||||
|
||||
has_footprint = any(
|
||||
db.query(model).filter(condition).first() is not None
|
||||
for model, condition in [
|
||||
(
|
||||
Test,
|
||||
(Test.created_by == user_id)
|
||||
| (Test.red_validated_by == user_id)
|
||||
| (Test.blue_validated_by == user_id)
|
||||
| (Test.remediation_assignee == user_id)
|
||||
| (Test.red_tech_assignee == user_id)
|
||||
| (Test.blue_tech_assignee == user_id)
|
||||
| (Test.red_reviewer_assignee == user_id)
|
||||
| (Test.blue_reviewer_assignee == user_id),
|
||||
),
|
||||
(Evidence, Evidence.uploaded_by == user_id),
|
||||
(AuditLog, AuditLog.user_id == user_id),
|
||||
(JiraLink, JiraLink.created_by == user_id),
|
||||
(Worklog, Worklog.user_id == user_id),
|
||||
(ProcedureSuggestion, (ProcedureSuggestion.submitted_by == user_id) | (ProcedureSuggestion.reviewed_by == user_id)),
|
||||
(TemplateSuggestion, (TemplateSuggestion.submitted_by == user_id) | (TemplateSuggestion.reviewed_by == user_id)),
|
||||
(TestRoundHistory, TestRoundHistory.reviewed_by == user_id),
|
||||
(EvaluationImport, EvaluationImport.imported_by == user_id),
|
||||
]
|
||||
)
|
||||
if has_footprint:
|
||||
raise BusinessRuleViolation(
|
||||
"This user has activity history (tests, evidence, worklogs, audit "
|
||||
"entries, or similar) and can't be permanently deleted — "
|
||||
"deactivate them instead to keep the audit trail intact."
|
||||
)
|
||||
|
||||
db.query(Notification).filter(Notification.user_id == user_id).delete()
|
||||
db.query(PasswordSetupToken).filter(PasswordSetupToken.user_id == user_id).delete()
|
||||
db.delete(user)
|
||||
|
||||
|
||||
def switch_active_role(db: Session, user: User, new_role: str) -> User:
|
||||
"""Swap *user*'s currently-active role with one from their extra_roles.
|
||||
|
||||
|
||||
@@ -0,0 +1,96 @@
|
||||
"""Admins can permanently delete users with no activity footprint —
|
||||
anyone with tests/evidence/worklogs/etc must be deactivated instead, to
|
||||
keep the audit trail intact.
|
||||
"""
|
||||
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def technique(client, auth_headers):
|
||||
response = client.post(
|
||||
"/api/v1/techniques",
|
||||
json={"mitre_id": "T1059", "name": "Test Technique"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
return response.json()
|
||||
|
||||
|
||||
def test_admin_can_delete_user_with_no_footprint(api, auth_headers):
|
||||
created = api(
|
||||
"post", "/api/v1/users", auth_headers,
|
||||
json={"full_name": "Disposable User", "email": "disposable@test.com", "role": "viewer"},
|
||||
)
|
||||
assert created.status_code == 201, created.text
|
||||
user_id = created.json()["id"]
|
||||
|
||||
resp = api("delete", f"/api/v1/users/{user_id}", auth_headers)
|
||||
assert resp.status_code == 204, resp.text
|
||||
|
||||
get_resp = api("get", f"/api/v1/users/{user_id}", auth_headers)
|
||||
assert get_resp.status_code == 404
|
||||
|
||||
|
||||
def test_admin_cannot_delete_own_account(api, auth_headers, admin_user):
|
||||
resp = api("delete", f"/api/v1/users/{admin_user.id}", auth_headers)
|
||||
assert resp.status_code == 400
|
||||
assert "own account" in resp.text.lower()
|
||||
|
||||
|
||||
def test_admin_cannot_delete_user_with_test_footprint(api, auth_headers, technique, red_lead_headers, red_lead_user, client):
|
||||
client.cookies.clear()
|
||||
test_resp = api(
|
||||
"post", "/api/v1/tests", red_lead_headers,
|
||||
json={
|
||||
"technique_id": technique["id"],
|
||||
"name": "Footprint Test",
|
||||
"description": "x",
|
||||
"platform": "windows",
|
||||
},
|
||||
)
|
||||
assert test_resp.status_code == 201, test_resp.text
|
||||
|
||||
resp = api("delete", f"/api/v1/users/{red_lead_user.id}", auth_headers)
|
||||
assert resp.status_code == 400
|
||||
assert "activity history" in resp.text.lower() or "deactivate" in resp.text.lower()
|
||||
|
||||
# The user must still be there — deactivating is the only allowed path.
|
||||
get_resp = api("get", f"/api/v1/users/{red_lead_user.id}", auth_headers)
|
||||
assert get_resp.status_code == 200
|
||||
|
||||
|
||||
def test_delete_user_requires_admin(api, red_lead_headers, red_lead_user):
|
||||
resp = api("delete", f"/api/v1/users/{red_lead_user.id}", red_lead_headers)
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
def test_delete_user_not_found(api, auth_headers):
|
||||
resp = api("delete", f"/api/v1/users/{uuid.uuid4()}", auth_headers)
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
def test_delete_cleans_up_notifications_and_tokens(api, auth_headers, db):
|
||||
created = api(
|
||||
"post", "/api/v1/users", auth_headers,
|
||||
json={"full_name": "Notifiable User", "email": "notifiable@test.com", "role": "viewer"},
|
||||
)
|
||||
user_id = uuid.UUID(created.json()["id"])
|
||||
|
||||
from app.models.notification import Notification
|
||||
from app.models.password_setup_token import PasswordSetupToken
|
||||
from app.services.notification_service import create_notification
|
||||
|
||||
create_notification(
|
||||
db, user_id=user_id, type="test", title="Hi", message="hi",
|
||||
entity_type="test", entity_id=uuid.uuid4(),
|
||||
)
|
||||
db.add(PasswordSetupToken(user_id=user_id, token="tok-abc", expires_at=__import__("datetime").datetime.utcnow()))
|
||||
db.commit()
|
||||
|
||||
resp = api("delete", f"/api/v1/users/{user_id}", auth_headers)
|
||||
assert resp.status_code == 204, resp.text
|
||||
|
||||
assert db.query(Notification).filter(Notification.user_id == user_id).count() == 0
|
||||
assert db.query(PasswordSetupToken).filter(PasswordSetupToken.user_id == user_id).count() == 0
|
||||
@@ -38,6 +38,11 @@ async function tryRefresh(): Promise<boolean> {
|
||||
}
|
||||
}
|
||||
|
||||
// Routes that are public — a 401 from a background call (e.g. the auth
|
||||
// provider's mount-time /auth/me check) must never bounce someone away
|
||||
// from these while they're using them logged out.
|
||||
const PUBLIC_PATHS = ["/login", "/set-password"];
|
||||
|
||||
// ── Response interceptor ────────────────────────────────────────────
|
||||
client.interceptors.response.use(
|
||||
(response) => response,
|
||||
@@ -58,8 +63,10 @@ client.interceptors.response.use(
|
||||
}
|
||||
}
|
||||
|
||||
// Refresh failed or not applicable — redirect to login
|
||||
if (window.location.pathname !== "/login") {
|
||||
// Refresh failed or not applicable — redirect to login, unless
|
||||
// we're already on a public page (e.g. /set-password) that a
|
||||
// logged-out visitor is legitimately using.
|
||||
if (!PUBLIC_PATHS.includes(window.location.pathname)) {
|
||||
window.location.href = "/login";
|
||||
}
|
||||
}
|
||||
|
||||
@@ -61,6 +61,13 @@ export async function updateUser(userId: string, payload: UserUpdatePayload): Pr
|
||||
return data;
|
||||
}
|
||||
|
||||
/** Permanently delete a user (admin only). Fails with a 400 if the user has
|
||||
* any activity footprint (tests, evidence, worklogs, audit entries, etc) —
|
||||
* deactivate those instead to preserve the audit trail. */
|
||||
export async function deleteUser(userId: string): Promise<void> {
|
||||
await client.delete(`/users/${userId}`);
|
||||
}
|
||||
|
||||
/** Email a one-time set-password link to a user (admin only). Also used for password resets. */
|
||||
export async function sendPasswordEmail(userId: string): Promise<{ detail: string }> {
|
||||
const { data } = await client.post<{ detail: string }>(`/users/${userId}/send-password-email`);
|
||||
|
||||
@@ -10,11 +10,13 @@ import {
|
||||
UserCheck,
|
||||
Edit,
|
||||
Mail,
|
||||
Trash2,
|
||||
} from "lucide-react";
|
||||
import {
|
||||
getUsers,
|
||||
createUser,
|
||||
updateUser,
|
||||
deleteUser,
|
||||
sendPasswordEmail,
|
||||
type UserOut,
|
||||
type UserCreatePayload,
|
||||
@@ -85,6 +87,26 @@ export default function UsersPage() {
|
||||
onError: (err: Error) => showToast("error", err.message),
|
||||
});
|
||||
|
||||
const deleteMutation = useMutation({
|
||||
mutationFn: deleteUser,
|
||||
onSuccess: () => {
|
||||
queryClient.invalidateQueries({ queryKey: ["users"] });
|
||||
showToast("success", "User deleted");
|
||||
},
|
||||
onError: (err: Error) => showToast("error", err.message),
|
||||
});
|
||||
|
||||
const handleDelete = (user: UserOut) => {
|
||||
if (
|
||||
window.confirm(
|
||||
`Permanently delete ${user.full_name || user.username}? This cannot be undone. ` +
|
||||
`Users with any activity (tests, evidence, worklogs, etc) can't be deleted — deactivate them instead.`,
|
||||
)
|
||||
) {
|
||||
deleteMutation.mutate(user.id);
|
||||
}
|
||||
};
|
||||
|
||||
const toggleUserActive = (user: UserOut) => {
|
||||
updateMutation.mutate({
|
||||
userId: user.id,
|
||||
@@ -240,6 +262,14 @@ export default function UsersPage() {
|
||||
<UserCheck className="h-4 w-4" />
|
||||
)}
|
||||
</button>
|
||||
<button
|
||||
onClick={() => handleDelete(user)}
|
||||
disabled={deleteMutation.isPending || user.id === currentUser?.id}
|
||||
className="rounded p-1.5 text-gray-400 hover:bg-red-900/50 hover:text-red-400 disabled:opacity-30 disabled:hover:bg-transparent disabled:hover:text-gray-400"
|
||||
title={user.id === currentUser?.id ? "You can't delete your own account" : "Delete permanently"}
|
||||
>
|
||||
<Trash2 className="h-4 w-4" />
|
||||
</button>
|
||||
</div>
|
||||
</td>
|
||||
</tr>
|
||||
|
||||
Reference in New Issue
Block a user