[feat] admin impersonate user (#3741)

2026-02-25 07:53:47 +01:00
parent 87b66c554a
commit 9be894da74
13 changed files with 359 additions and 19 deletions
@@ -2047,3 +2047,193 @@ async def test_api_update_user_labels(http_client: AsyncClient):
        """string does not match regex "([A-Za-z0-9 ._-]{1,100}$)"""
        in data["detail"][0]["msg"]
    )
+
+
+@pytest.mark.anyio
+async def test_impersonate_user_success(http_client: AsyncClient, admin_user: User):
+    tiny_id = shortuuid.uuid()[:8]
+    user_id = uuid4().hex
+    account = Account(
+        id=user_id,
+        username=f"u_{tiny_id}",
+        email=f"u_{tiny_id}@lnbits.com",
+    )
+    account.hash_password("secret1234")
+    await create_user_account(account)
+
+    # Login as admin to get access token
+    response = await http_client.post(
+        "/api/v1/auth", json={"username": admin_user.username, "password": "secret1234"}
+    )
+    assert response.status_code == 200
+    admin_token = response.json()["access_token"]
+
+    # Impersonate the user
+    response = await http_client.post(
+        "/api/v1/auth/impersonate",
+        json={"usr": user_id},
+        headers={"Authorization": f"Bearer {admin_token}"},
+    )
+    assert response.status_code == 200
+    data = response.json()
+    assert "access_token" in data
+    assert data["token_type"] == "bearer"
+
+    # Check impersonation cookies
+    assert "cookie_access_token" in response.cookies
+    assert "admin_access_token" in response.cookies
+    assert response.cookies.get("is_lnbits_user_impersonated") == "true"
+
+    response = await http_client.delete("/api/v1/auth/impersonate")
+    assert "cookie_access_token" in response.cookies
+    assert response.cookies.get("cookie_access_token") == admin_token
+    assert "admin_access_token" not in response.cookies
+    assert "is_lnbits_user_impersonated" not in response.cookies
+
+
+@pytest.mark.anyio
+async def test_impersonate_user_no_cookie(http_client: AsyncClient, admin_user: User):
+    response = await http_client.post(
+        "/api/v1/auth", json={"username": admin_user.username, "password": "secret1234"}
+    )
+    admin_token = response.json()["access_token"]
+    user_id = uuid4().hex
+    http_client.cookies.clear()
+    response = await http_client.post(
+        "/api/v1/auth/impersonate",
+        json={"usr": user_id},
+        headers={"Authorization": f"Bearer {admin_token}"},
+    )
+
+    assert response.status_code == 401
+    assert response.json()["detail"] == "Only cookie based impersonation is allowed."
+
+
+@pytest.mark.anyio
+async def test_impersonate_user_self(http_client: AsyncClient, admin_user: User):
+    # Admin tries to impersonate themselves
+    response = await http_client.post(
+        "/api/v1/auth", json={"username": admin_user.username, "password": "secret1234"}
+    )
+    admin_token = response.json()["access_token"]
+    response = await http_client.post(
+        "/api/v1/auth/impersonate",
+        json={"usr": admin_user.id},
+        headers={"Authorization": f"Bearer {admin_token}"},
+    )
+
+    assert response.status_code == 403
+    assert response.json()["detail"] == "You cannot impersonate yourself."
+
+
+@pytest.mark.anyio
+async def test_impersonate_user_admin_target(
+    http_client: AsyncClient, admin_user: User, settings: Settings
+):
+    other_admin_id = uuid4().hex
+    tiny_id = shortuuid.uuid()[:8]
+    account = Account(
+        id=other_admin_id, username=f"u_{tiny_id}", email=f"u_{tiny_id}@lnbits.com"
+    )
+    account.hash_password("secret1234")
+
+    await create_user_account(account)
+    settings.lnbits_admin_users.append(other_admin_id)
+
+    response = await http_client.post(
+        "/api/v1/auth", json={"username": admin_user.username, "password": "secret1234"}
+    )
+    admin_token = response.json()["access_token"]
+    response = await http_client.post(
+        "/api/v1/auth/impersonate",
+        json={"usr": other_admin_id},
+        headers={"Authorization": f"Bearer {admin_token}"},
+    )
+    assert response.status_code == 403
+    assert response.json()["detail"] == "You cannot impersonate another admin user."
+
+
+@pytest.mark.anyio
+async def test_impersonate_user_nonexistent(
+    http_client: AsyncClient, admin_user: User, settings: Settings
+):
+    http_client.cookies.clear()
+    response = await http_client.post(
+        "/api/v1/auth", json={"username": admin_user.username, "password": "secret1234"}
+    )
+    admin_token = response.json()["access_token"]
+    fake_id = "deadbeef"
+    response = await http_client.post(
+        "/api/v1/auth/impersonate",
+        json={"usr": fake_id},
+        headers={"Authorization": f"Bearer {admin_token}"},
+    )
+    assert response.status_code == 401
+    assert response.json()["detail"] == "User ID does not exist."
+
+
+@pytest.mark.anyio
+async def test_impersonate_user_invalid_data(
+    http_client: AsyncClient, admin_user: User
+):
+    response = await http_client.post(
+        "/api/v1/auth", json={"username": admin_user.username, "password": "secret1234"}
+    )
+    admin_token = response.json()["access_token"]
+    response = await http_client.post(
+        "/api/v1/auth/impersonate",
+        json={"usr": None},
+        headers={"Authorization": f"Bearer {admin_token}"},
+    )
+
+    assert response.status_code == 400
+    assert "type_error.none.not_allowed" in str(response.json()["detail"])
+
+
+@pytest.mark.anyio
+async def test_impersonate_user_missing_usr_field(
+    http_client: AsyncClient, admin_user: User
+):
+    response = await http_client.post(
+        "/api/v1/auth", json={"username": admin_user.username, "password": "secret1234"}
+    )
+    admin_token = response.json()["access_token"]
+    response = await http_client.post(
+        "/api/v1/auth/impersonate",
+        json={},
+        headers={"Authorization": f"Bearer {admin_token}"},
+    )
+    assert response.status_code == 400
+    assert "value_error.missing" in str(response.json()["detail"])
+
+
+@pytest.mark.anyio
+async def test_impersonate_user_by_non_admin(http_client: AsyncClient, user_alan: User):
+    response = await http_client.post(
+        "/api/v1/auth", json={"username": user_alan.username, "password": "secret1234"}
+    )
+    alan_token = response.json()["access_token"]
+    response = await http_client.post(
+        "/api/v1/auth/impersonate",
+        json={},
+        headers={"Authorization": f"Bearer {alan_token}"},
+    )
+    assert response.status_code == 403
+    assert response.json()["detail"] == "User not authorized. No admin privileges."
+
+
+@pytest.mark.anyio
+async def test_stop_impersonate_user_by_non_admin(
+    http_client: AsyncClient, user_alan: User
+):
+    response = await http_client.post(
+        "/api/v1/auth", json={"username": user_alan.username, "password": "secret1234"}
+    )
+
+    response = await http_client.delete("/api/v1/auth/impersonate")
+
+    assert response.status_code == 401
+    assert (
+        response.json()["detail"]
+        == "No admin access token found to stop impersonation."
+    )
@@ -14,9 +14,9 @@ from lnbits.core.crud import (
    delete_account,
    get_account_by_username,
    get_payment,
+    get_user,
    update_payment,
 )
-from lnbits.core.crud.users import get_user_from_account
 from lnbits.core.models import Account, CreateInvoice, PaymentState, User
 from lnbits.core.models.users import UpdateSuperuserPassword
 from lnbits.core.services import create_user_account, update_wallet_balance
@@ -114,10 +114,10 @@ async def user_alan():

@pytest.fixture(scope="session")
 async def admin_user():
-    username = "admin"
-    account = await get_account_by_username(username)
-    if account:
-        return await get_user_from_account(account)
+    username = "admin_" + uuid4().hex[:8]
+    user = await get_user(ADMIN_USER_ID)
+    if user:
+        return user

    account = Account(
        id=ADMIN_USER_ID,