test_users_slim_1894.py 7.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180
  1. """GET /api/v1/users/slim -- the id -> username mapping for API clients (#1894).
  2. An API key could already read global archive stats and filter them by
  3. ``created_by_id`` (for API-keyed requests the permission deps return None as
  4. ``current_user``, so the ``stats:filter_by_user`` guard short-circuits), but
  5. had no way to discover which id belonged to whom: the full listing is gated on
  6. ``users:read``, which is unmapped in the API-key scope allowlist and therefore
  7. administrative.
  8. The slim listing closes that gap without handing keys the full user objects.
  9. These tests pin both halves: that it answers for a key, and that it stays
  10. narrow while the full listing stays admin-only.
  11. """
  12. import pytest
  13. from httpx import AsyncClient
  14. from sqlalchemy import select
  15. from backend.app.core.auth import generate_api_key
  16. from backend.app.models.api_key import APIKey
  17. from backend.app.models.group import Group
  18. from backend.app.models.user import User
  19. async def _setup_and_login(async_client: AsyncClient) -> str:
  20. await async_client.post(
  21. "/api/v1/auth/setup",
  22. json={"auth_enabled": True, "admin_username": "slimadmin", "admin_password": "SlimPass1!"},
  23. )
  24. login = await async_client.post(
  25. "/api/v1/auth/login",
  26. json={"username": "slimadmin", "password": "SlimPass1!"},
  27. )
  28. return login.json()["access_token"]
  29. async def _add_key(db_session, *, user_id: int | None = None, **scopes) -> str:
  30. full_key, key_hash, key_prefix = generate_api_key()
  31. db_session.add(
  32. APIKey(name="slim-test", key_hash=key_hash, key_prefix=key_prefix, enabled=True, user_id=user_id, **scopes)
  33. )
  34. await db_session.commit()
  35. return full_key
  36. async def _add_user(db_session, username: str, **kwargs) -> User:
  37. from backend.app.core.auth import get_password_hash
  38. user = User(
  39. username=username,
  40. password_hash=get_password_hash("Whatever1!"),
  41. email=f"{username}@example.invalid",
  42. role="user",
  43. is_active=True,
  44. **kwargs,
  45. )
  46. db_session.add(user)
  47. await db_session.commit()
  48. return user
  49. @pytest.mark.asyncio
  50. @pytest.mark.integration
  51. async def test_slim_returns_only_id_and_username(async_client: AsyncClient, db_session):
  52. """The response shape is the contract -- no emails, roles, or permissions."""
  53. token = await _setup_and_login(async_client)
  54. await _add_user(db_session, "bob")
  55. response = await async_client.get("/api/v1/users/slim", headers={"Authorization": f"Bearer {token}"})
  56. assert response.status_code == 200
  57. rows = response.json()
  58. assert rows, "expected at least the admin created by setup"
  59. for row in rows:
  60. assert set(row) == {"id", "username"}
  61. assert "bob" in [row["username"] for row in rows]
  62. @pytest.mark.asyncio
  63. @pytest.mark.integration
  64. async def test_slim_is_reachable_with_an_api_key(async_client: AsyncClient, db_session):
  65. """The point of the issue: a key can resolve the ids it already filters on."""
  66. await _setup_and_login(async_client)
  67. owner = (await db_session.execute(select(User).where(User.username == "slimadmin"))).scalar_one()
  68. full_key = await _add_key(db_session, user_id=owner.id, can_read_status=True)
  69. header = await async_client.get("/api/v1/users/slim", headers={"X-API-Key": full_key})
  70. bearer = await async_client.get("/api/v1/users/slim", headers={"Authorization": f"Bearer {full_key}"})
  71. assert header.status_code == 200
  72. assert bearer.status_code == 200
  73. assert {row["username"] for row in header.json()} == {"slimadmin"}
  74. @pytest.mark.asyncio
  75. @pytest.mark.integration
  76. async def test_slim_needs_can_read_status(async_client: AsyncClient, db_session):
  77. """A key without the read scope gets nothing, same as any other read route."""
  78. await _setup_and_login(async_client)
  79. owner = (await db_session.execute(select(User).where(User.username == "slimadmin"))).scalar_one()
  80. full_key = await _add_key(db_session, user_id=owner.id, can_read_status=False)
  81. response = await async_client.get("/api/v1/users/slim", headers={"X-API-Key": full_key})
  82. assert response.status_code == 403
  83. @pytest.mark.asyncio
  84. @pytest.mark.integration
  85. async def test_full_listing_stays_admin_only_for_api_keys(async_client: AsyncClient, db_session):
  86. """Regression guard: widening the slim route must not widen the full one.
  87. ``users:read`` returns emails, group membership and the complete permission
  88. set for every account. It has to stay unmapped in the scope allowlist.
  89. """
  90. await _setup_and_login(async_client)
  91. owner = (await db_session.execute(select(User).where(User.username == "slimadmin"))).scalar_one()
  92. full_key = await _add_key(db_session, user_id=owner.id, can_read_status=True)
  93. response = await async_client.get("/api/v1/users", headers={"X-API-Key": full_key})
  94. assert response.status_code == 403
  95. assert "administrative" in response.json()["detail"]
  96. @pytest.mark.asyncio
  97. @pytest.mark.integration
  98. async def test_slim_is_not_parsed_as_a_user_id(async_client: AsyncClient, db_session):
  99. """Route ordering. Declared after /{user_id}, "slim" would 422 as an int."""
  100. token = await _setup_and_login(async_client)
  101. response = await async_client.get("/api/v1/users/slim", headers={"Authorization": f"Bearer {token}"})
  102. assert response.status_code != 422
  103. @pytest.mark.asyncio
  104. @pytest.mark.integration
  105. async def test_a_group_with_only_users_read_still_reaches_slim(async_client: AsyncClient, db_session):
  106. """``users:read`` is strictly broader, so it must pass the any-of gate.
  107. Without this, every existing custom group holding ``users:read`` would need
  108. a permission backfill before the frontend could ever move to this route.
  109. """
  110. await _setup_and_login(async_client)
  111. group = Group(name="readers", description="t", permissions=["users:read"], is_system=False)
  112. db_session.add(group)
  113. await db_session.flush()
  114. await _add_user(db_session, "reader", groups=[group])
  115. login = await async_client.post(
  116. "/api/v1/auth/login",
  117. json={"username": "reader", "password": "Whatever1!"},
  118. )
  119. token = login.json()["access_token"]
  120. response = await async_client.get("/api/v1/users/slim", headers={"Authorization": f"Bearer {token}"})
  121. assert response.status_code == 200
  122. @pytest.mark.asyncio
  123. @pytest.mark.integration
  124. async def test_a_group_with_only_slim_cannot_read_the_full_listing(async_client: AsyncClient, db_session):
  125. """The narrow grant has to actually be narrower for JWT users too."""
  126. await _setup_and_login(async_client)
  127. group = Group(name="slim-only", description="t", permissions=["users:read_slim"], is_system=False)
  128. db_session.add(group)
  129. await db_session.flush()
  130. await _add_user(db_session, "slimonly", groups=[group])
  131. login = await async_client.post(
  132. "/api/v1/auth/login",
  133. json={"username": "slimonly", "password": "Whatever1!"},
  134. )
  135. token = login.json()["access_token"]
  136. headers = {"Authorization": f"Bearer {token}"}
  137. assert (await async_client.get("/api/v1/users/slim", headers=headers)).status_code == 200
  138. assert (await async_client.get("/api/v1/users", headers=headers)).status_code == 403