sakshamandClaude Sonnet 5
4ee38b6241
Logs a distinct db_viewer_accessed event (not just the earlier session_created "requested" event) when an admin's browser actually completes the hand-off into Adminer. Adds a password-confirmed account-deletion box to Settings, relying on the existing ON DELETE CASCADE foreign keys to clean up everything the account owns. Adds an admin-only "require password change" flag that get_current_user enforces server-side (403 on everything except /auth/me, /auth/change-password, /auth/logout) - meant for handing a demo account to someone with a known sample password. Co-Authored-By: Claude Sonnet 5 <[email protected]>
261 lines
9.0 KiB
Python
261 lines
9.0 KiB
Python
"""Account deletion (DELETE /auth/me) and admin-forced password change
|
|
(POST /admin/users/require-password-change -> POST /auth/change-password,
|
|
enforced by app.auth.dependencies.get_current_user)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.enums import SecurityEventType
|
|
from app.repositories.user_repository import UserRepository
|
|
from app.repositories.user_security_event_repository import UserSecurityEventRepository
|
|
|
|
|
|
def _unique_email() -> str:
|
|
return f"user-{uuid.uuid4().hex[:12]}@example.com"
|
|
|
|
|
|
def _register_and_login(client: TestClient, password: str = "correct-horse-1") -> dict:
|
|
email = _unique_email()
|
|
client.post(
|
|
"/api/v1/auth/register",
|
|
json={"email": email, "password": password, "display_name": "T"},
|
|
)
|
|
tokens = client.post("/api/v1/auth/login", json={"email": email, "password": password}).json()
|
|
return {"email": email, "headers": {"Authorization": f"Bearer {tokens['access_token']}"}}
|
|
|
|
|
|
async def _register_admin_and_login(client: TestClient, db_session: AsyncSession) -> dict[str, str]:
|
|
email = f"admin-{uuid.uuid4().hex[:12]}@example.com"
|
|
client.post(
|
|
"/api/v1/auth/register",
|
|
json={"email": email, "password": "correct-horse-1", "display_name": "Admin"},
|
|
)
|
|
user = await UserRepository(db_session).get_by_email(email)
|
|
user.is_admin = True
|
|
await db_session.commit()
|
|
tokens = client.post(
|
|
"/api/v1/auth/login", json={"email": email, "password": "correct-horse-1"}
|
|
).json()
|
|
return {"Authorization": f"Bearer {tokens['access_token']}"}
|
|
|
|
|
|
# --- DELETE /auth/me -------------------------------------------------------
|
|
|
|
|
|
def test_delete_account_requires_auth(client: TestClient):
|
|
resp = client.request(
|
|
"DELETE", "/api/v1/auth/me", json={"password": "whatever"}
|
|
)
|
|
assert resp.status_code == 401
|
|
|
|
|
|
def test_delete_account_rejects_wrong_password(client: TestClient):
|
|
account = _register_and_login(client)
|
|
resp = client.request(
|
|
"DELETE",
|
|
"/api/v1/auth/me",
|
|
json={"password": "not-the-right-password"},
|
|
headers=account["headers"],
|
|
)
|
|
assert resp.status_code == 401
|
|
|
|
|
|
def test_delete_account_succeeds_with_correct_password_and_logs_out_the_account(
|
|
client: TestClient,
|
|
):
|
|
account = _register_and_login(client)
|
|
resp = client.request(
|
|
"DELETE",
|
|
"/api/v1/auth/me",
|
|
json={"password": "correct-horse-1"},
|
|
headers=account["headers"],
|
|
)
|
|
assert resp.status_code == 204
|
|
|
|
# The account is gone - the same token no longer resolves to anyone.
|
|
me_resp = client.get("/api/v1/auth/me", headers=account["headers"])
|
|
assert me_resp.status_code == 401
|
|
|
|
# And a fresh login attempt with the same credentials fails too.
|
|
login_resp = client.post(
|
|
"/api/v1/auth/login",
|
|
json={"email": account["email"], "password": "correct-horse-1"},
|
|
)
|
|
assert login_resp.status_code == 401
|
|
|
|
|
|
async def test_delete_account_cascades_to_owned_data(client: TestClient, db_session: AsyncSession):
|
|
"""A company created by the account must be gone too (ON DELETE CASCADE),
|
|
not just the user row."""
|
|
account = _register_and_login(client)
|
|
create_resp = client.post(
|
|
"/api/v1/companies",
|
|
json={"name": f"Co-{uuid.uuid4().hex[:8]}", "official_website": None},
|
|
headers=account["headers"],
|
|
)
|
|
assert create_resp.status_code == 201
|
|
company_id = create_resp.json()["id"]
|
|
|
|
client.request(
|
|
"DELETE", "/api/v1/auth/me", json={"password": "correct-horse-1"}, headers=account["headers"]
|
|
)
|
|
|
|
from sqlalchemy import text
|
|
|
|
row = await db_session.execute(
|
|
text("SELECT 1 FROM companies WHERE id = :id"), {"id": company_id}
|
|
)
|
|
assert row.first() is None
|
|
|
|
|
|
# --- POST /admin/users/require-password-change -----------------------------
|
|
|
|
|
|
def test_require_password_change_requires_admin(client: TestClient):
|
|
account = _register_and_login(client)
|
|
resp = client.post(
|
|
"/api/v1/admin/users/require-password-change",
|
|
json={"email": account["email"]},
|
|
headers=account["headers"],
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
|
|
async def test_require_password_change_rejects_unknown_email(
|
|
client: TestClient, db_session: AsyncSession
|
|
):
|
|
headers = await _register_admin_and_login(client, db_session)
|
|
resp = client.post(
|
|
"/api/v1/admin/users/require-password-change",
|
|
json={"email": "[email protected]"},
|
|
headers=headers,
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
async def test_admin_flags_an_account_and_it_gets_logged_to_the_targets_activity(
|
|
client: TestClient, db_session: AsyncSession
|
|
):
|
|
"""Checked directly against the DB, not via GET /auth/security-events -
|
|
that endpoint isn't in the must-change-password exempt list, so the
|
|
just-flagged account can't reach it until they change their password
|
|
(see test_flagged_account_is_blocked_from_other_endpoints)."""
|
|
admin_headers = await _register_admin_and_login(client, db_session)
|
|
account = _register_and_login(client)
|
|
|
|
resp = client.post(
|
|
"/api/v1/admin/users/require-password-change",
|
|
json={"email": account["email"]},
|
|
headers=admin_headers,
|
|
)
|
|
assert resp.status_code == 204
|
|
|
|
user = await UserRepository(db_session).get_by_email(account["email"])
|
|
assert user.must_change_password is True
|
|
events = await UserSecurityEventRepository(db_session).list_for_user(user.id)
|
|
assert any(e.event_type == SecurityEventType.PASSWORD_CHANGE_REQUIRED for e in events)
|
|
|
|
|
|
# --- Enforcement gate: everything except me/change-password/logout 403s ----
|
|
|
|
|
|
async def test_flagged_account_is_blocked_from_other_endpoints(
|
|
client: TestClient, db_session: AsyncSession
|
|
):
|
|
admin_headers = await _register_admin_and_login(client, db_session)
|
|
account = _register_and_login(client)
|
|
client.post(
|
|
"/api/v1/admin/users/require-password-change",
|
|
json={"email": account["email"]},
|
|
headers=admin_headers,
|
|
)
|
|
|
|
blocked = client.get("/api/v1/companies", headers=account["headers"])
|
|
assert blocked.status_code == 403
|
|
|
|
still_ok = client.get("/api/v1/auth/me", headers=account["headers"])
|
|
assert still_ok.status_code == 200
|
|
assert still_ok.json()["must_change_password"] is True
|
|
|
|
|
|
async def test_change_password_clears_the_flag_and_unblocks_the_account(
|
|
client: TestClient, db_session: AsyncSession
|
|
):
|
|
admin_headers = await _register_admin_and_login(client, db_session)
|
|
account = _register_and_login(client)
|
|
client.post(
|
|
"/api/v1/admin/users/require-password-change",
|
|
json={"email": account["email"]},
|
|
headers=admin_headers,
|
|
)
|
|
|
|
change_resp = client.post(
|
|
"/api/v1/auth/change-password",
|
|
json={"current_password": "correct-horse-1", "new_password": "brand-new-horse-2"},
|
|
headers=account["headers"],
|
|
)
|
|
assert change_resp.status_code == 200
|
|
new_tokens = change_resp.json()
|
|
new_headers = {"Authorization": f"Bearer {new_tokens['access_token']}"}
|
|
|
|
me_resp = client.get("/api/v1/auth/me", headers=new_headers)
|
|
assert me_resp.status_code == 200
|
|
assert me_resp.json()["must_change_password"] is False
|
|
|
|
unblocked = client.get("/api/v1/companies", headers=new_headers)
|
|
assert unblocked.status_code == 200
|
|
|
|
# The new password actually works on a fresh login.
|
|
login_resp = client.post(
|
|
"/api/v1/auth/login",
|
|
json={"email": account["email"], "password": "brand-new-horse-2"},
|
|
)
|
|
assert login_resp.status_code == 200
|
|
|
|
|
|
def test_change_password_rejects_wrong_current_password(client: TestClient):
|
|
account = _register_and_login(client)
|
|
resp = client.post(
|
|
"/api/v1/auth/change-password",
|
|
json={"current_password": "totally-wrong", "new_password": "brand-new-horse-2"},
|
|
headers=account["headers"],
|
|
)
|
|
assert resp.status_code == 401
|
|
|
|
|
|
def test_change_password_rejects_reusing_the_current_password(client: TestClient):
|
|
account = _register_and_login(client)
|
|
resp = client.post(
|
|
"/api/v1/auth/change-password",
|
|
json={"current_password": "correct-horse-1", "new_password": "correct-horse-1"},
|
|
headers=account["headers"],
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_change_password_revokes_the_old_refresh_token(client: TestClient):
|
|
email = _unique_email()
|
|
client.post(
|
|
"/api/v1/auth/register",
|
|
json={"email": email, "password": "correct-horse-1", "display_name": "T"},
|
|
)
|
|
tokens = client.post(
|
|
"/api/v1/auth/login", json={"email": email, "password": "correct-horse-1"}
|
|
).json()
|
|
headers = {"Authorization": f"Bearer {tokens['access_token']}"}
|
|
|
|
client.post(
|
|
"/api/v1/auth/change-password",
|
|
json={"current_password": "correct-horse-1", "new_password": "brand-new-horse-2"},
|
|
headers=headers,
|
|
)
|
|
|
|
stale_refresh_resp = client.post(
|
|
"/api/v1/auth/refresh", json={"refresh_token": tokens["refresh_token"]}
|
|
)
|
|
assert stale_refresh_resp.status_code == 401
|