mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-10-22 03:49:18 +02:00
Let API clients resolve user ids to names (#1894)
Archives, the queue and statistics report ownership as a numeric created_by_id, and statistics accept it as a filter, but nothing let an API key discover whose id was whose -- the only user listing returns emails, roles, group membership and full permission sets, so it is administrative and rejects keys. Add GET /users/slim returning id + username only, gated on a new users:read_slim permission mapped to can_read_status. That grants no data a key could not already reach: for API-keyed requests the permission deps return None as current_user, so the stats:filter_by_user guard short-circuits and ?created_by_id=N is already honoured for every N. What was missing was the ability to address the filter, not permission to use it. The full listing stays unmapped = admin-only. Also fix /auth/me, which answered an API key with a synthetic administrator: id 0, role admin, is_admin true and every permission in the enum. A key cannot reach an administrative route at all, so clients building their UI from that response rendered actions that 403 on use. It now reports the key owner's identity, is_admin false, and the permissions the key's scopes actually admit. Ownerless legacy keys keep id 0 but no longer claim admin. --- Source user names from the slim listing where only names are needed (#1894) Stats filter-by-user, the Archives print log filter, the File Manager username autocomplete, the camera-token owner column and the Finance member picker all render nothing but a username, but all of them read the full user listing, which is gated on the admin-level users:read. An operator granted stats:filter_by_user but not users:read got an empty filter with no indication why. Point them at /users/slim under a separate react-query key, since the full listing shares the 'users' key and the two shapes would clobber each other in the cache.
This commit is contained in:
maziggy
@@ -0,0 +1,240 @@
|
||||
"""An API key must not out-rank the user it belongs to (#1894 follow-on).
|
||||
|
||||
``_check_apikey_permissions`` gated purely on the scope flags stored on the key
|
||||
row and never looked at the owner. Scope flags are chosen at creation time by
|
||||
whoever holds ``api_keys:create`` -- admin-only in the default groups, but a
|
||||
custom group can grant it -- so a user with, say, queue permissions could mint
|
||||
themselves a key with ``can_control_printer`` and stop other people's prints
|
||||
through it. Deactivating that user did not help either: their keys kept working
|
||||
with full scope authority, because nothing re-checked the owner.
|
||||
|
||||
The gate now narrows the scope flags to what the owner may do. Two cases must
|
||||
NOT be conflated, and each has a test below:
|
||||
|
||||
- ``user_id IS NULL`` -- legacy key from before per-user ownership. No owner
|
||||
exists to narrow against, so the flags stand alone and the key keeps working.
|
||||
- ``user_id`` set but the row is gone or deactivated -- the key's authority came
|
||||
from a user who has none. Fails closed.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
from sqlalchemy import select
|
||||
|
||||
from backend.app.core.auth import generate_api_key, get_password_hash
|
||||
from backend.app.models.api_key import APIKey
|
||||
from backend.app.models.group import Group
|
||||
from backend.app.models.user import User
|
||||
|
||||
# A route gated on PRINTERS_READ (can_read_status) and one gated on
|
||||
# PRINTERS_CONTROL (can_control_printer). Both scope flags are set on every key
|
||||
# built below, so any denial comes from the owner check rather than the flags.
|
||||
READ_ROUTE = "/api/v1/printers/"
|
||||
CONTROL_ROUTE = "/api/v1/printers/1/print/stop"
|
||||
|
||||
|
||||
async def _setup(async_client: AsyncClient) -> None:
|
||||
await async_client.post(
|
||||
"/api/v1/auth/setup",
|
||||
json={"auth_enabled": True, "admin_username": "owneradmin", "admin_password": "OwnerPass1!"},
|
||||
)
|
||||
|
||||
|
||||
async def _key_for(db_session, owner: User | None, **scopes) -> str:
|
||||
defaults = {"can_read_status": True, "can_control_printer": True, "can_queue": True}
|
||||
defaults.update(scopes)
|
||||
full_key, key_hash, key_prefix = generate_api_key()
|
||||
db_session.add(
|
||||
APIKey(
|
||||
name=f"key-{owner.username if owner else 'legacy'}",
|
||||
key_hash=key_hash,
|
||||
key_prefix=key_prefix,
|
||||
enabled=True,
|
||||
user_id=owner.id if owner else None,
|
||||
**defaults,
|
||||
)
|
||||
)
|
||||
await db_session.commit()
|
||||
return full_key
|
||||
|
||||
|
||||
async def _user(db_session, username: str, permissions: list[str], *, is_active: bool = True) -> User:
|
||||
group = Group(name=f"grp-{username}", description="t", permissions=permissions, is_system=False)
|
||||
db_session.add(group)
|
||||
await db_session.flush()
|
||||
user = User(
|
||||
username=username,
|
||||
password_hash=get_password_hash("Whatever1!"), # noqa: S106
|
||||
role="user",
|
||||
is_active=is_active,
|
||||
groups=[group],
|
||||
)
|
||||
db_session.add(user)
|
||||
await db_session.commit()
|
||||
return user
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_admin_owned_key_keeps_full_scope_authority(async_client: AsyncClient, db_session):
|
||||
"""The common case must not regress -- almost every key is admin-owned."""
|
||||
await _setup(async_client)
|
||||
admin = (await db_session.execute(select(User).where(User.username == "owneradmin"))).scalar_one()
|
||||
key = await _key_for(db_session, admin)
|
||||
|
||||
response = await async_client.get(READ_ROUTE, headers={"X-API-Key": key})
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_legacy_ownerless_key_still_works(async_client: AsyncClient, db_session):
|
||||
"""No owner to narrow against is not the same as a failed owner lookup."""
|
||||
await _setup(async_client)
|
||||
key = await _key_for(db_session, None)
|
||||
|
||||
response = await async_client.get(READ_ROUTE, headers={"X-API-Key": key})
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_key_cannot_exceed_its_owners_permissions(async_client: AsyncClient, db_session):
|
||||
"""The escalation: control flags ticked, owner who may not control."""
|
||||
await _setup(async_client)
|
||||
owner = await _user(db_session, "readonly", ["printers:read"])
|
||||
key = await _key_for(db_session, owner)
|
||||
|
||||
allowed = await async_client.get(READ_ROUTE, headers={"X-API-Key": key})
|
||||
denied = await async_client.post(CONTROL_ROUTE, headers={"X-API-Key": key})
|
||||
|
||||
assert allowed.status_code == 200
|
||||
assert denied.status_code == 403
|
||||
assert "owner does not have" in denied.json()["detail"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_deactivating_the_owner_disables_the_key(async_client: AsyncClient, db_session):
|
||||
"""Previously the key kept working -- nothing re-checked the owner."""
|
||||
await _setup(async_client)
|
||||
owner = await _user(db_session, "gone", ["printers:read"], is_active=False)
|
||||
key = await _key_for(db_session, owner)
|
||||
|
||||
response = await async_client.get(READ_ROUTE, headers={"X-API-Key": key})
|
||||
|
||||
assert response.status_code == 403
|
||||
assert "deactivated" in response.json()["detail"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_a_deleted_owner_does_not_fall_back_to_anonymous(async_client: AsyncClient, db_session):
|
||||
"""The dangling-row case fails closed rather than reverting to flags-only.
|
||||
|
||||
CASCADE should prevent this, but "should" is not a gate -- if the row is
|
||||
ever orphaned the key must not silently regain full scope authority.
|
||||
"""
|
||||
await _setup(async_client)
|
||||
owner = await _user(db_session, "doomed", ["printers:read"])
|
||||
key = await _key_for(db_session, owner)
|
||||
api_key = (await db_session.execute(select(APIKey).where(APIKey.user_id == owner.id))).scalar_one()
|
||||
api_key.user_id = 999999 # owner row that does not exist
|
||||
await db_session.commit()
|
||||
|
||||
response = await async_client.get(READ_ROUTE, headers={"X-API-Key": key})
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_bearer_path_is_gated_the_same_as_the_header(async_client: AsyncClient, db_session):
|
||||
"""Both credential paths run the same gate; only one was ever tested."""
|
||||
await _setup(async_client)
|
||||
owner = await _user(db_session, "bearer-readonly", ["printers:read"])
|
||||
key = await _key_for(db_session, owner)
|
||||
|
||||
denied = await async_client.post(CONTROL_ROUTE, headers={"Authorization": f"Bearer {key}"})
|
||||
|
||||
assert denied.status_code == 403
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_webhook_routes_are_not_a_way_around_the_owner_check(async_client: AsyncClient, db_session):
|
||||
"""/webhook/* reaches its scope flags by a different route than the rest.
|
||||
|
||||
It gates on ``check_permission``, not ``_check_apikey_permissions``, so it
|
||||
does not inherit the owner narrowing for free. If it is missed, the same key
|
||||
that is refused on /printers/{id}/print/stop simply stops the print here
|
||||
instead, and the whole gate is decorative.
|
||||
"""
|
||||
await _setup(async_client)
|
||||
owner = await _user(db_session, "webhook-readonly", ["printers:read"])
|
||||
key = await _key_for(db_session, owner)
|
||||
|
||||
denied = await async_client.post("/api/v1/webhook/printer/1/stop", headers={"X-API-Key": key})
|
||||
|
||||
assert denied.status_code == 403
|
||||
assert "owner does not have" in denied.json()["detail"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_webhook_still_works_for_a_permitted_owner(async_client: AsyncClient, db_session):
|
||||
"""The narrowing must not simply break every webhook caller."""
|
||||
await _setup(async_client)
|
||||
owner = await _user(db_session, "webhook-operator", ["printers:read", "printers:control"])
|
||||
key = await _key_for(db_session, owner)
|
||||
|
||||
response = await async_client.post("/api/v1/webhook/printer/1/stop", headers={"X-API-Key": key})
|
||||
|
||||
# There is no connected printer 1, so the handler itself fails. What
|
||||
# matters is that the request got that far: neither gate rejected it.
|
||||
assert response.status_code != 403
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_webhook_rejects_a_deactivated_owner(async_client: AsyncClient, db_session):
|
||||
"""Fail-closed applies on this path too."""
|
||||
await _setup(async_client)
|
||||
owner = await _user(db_session, "webhook-gone", ["printers:read", "printers:control"], is_active=False)
|
||||
key = await _key_for(db_session, owner)
|
||||
|
||||
response = await async_client.post("/api/v1/webhook/printer/1/stop", headers={"X-API-Key": key})
|
||||
|
||||
assert response.status_code == 403
|
||||
assert "deactivated" in response.json()["detail"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_me_reports_the_narrowed_set(async_client: AsyncClient, db_session):
|
||||
"""/auth/me and the gate must agree, including about the owner."""
|
||||
await _setup(async_client)
|
||||
owner = await _user(db_session, "narrow", ["printers:read"])
|
||||
key = await _key_for(db_session, owner)
|
||||
|
||||
result = (await async_client.get("/api/v1/auth/me", headers={"X-API-Key": key})).json()
|
||||
|
||||
assert result["id"] == owner.id
|
||||
assert "printers:read" in result["permissions"]
|
||||
# can_control_printer is ticked on the key, but the owner cannot control.
|
||||
assert "printers:control" not in result["permissions"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_me_is_rejected_once_the_owner_is_deactivated(async_client: AsyncClient, db_session):
|
||||
"""A dead key identifies as nothing, rather than as an anonymous key."""
|
||||
await _setup(async_client)
|
||||
owner = await _user(db_session, "me-gone", ["printers:read"], is_active=False)
|
||||
key = await _key_for(db_session, owner)
|
||||
|
||||
response = await async_client.get("/api/v1/auth/me", headers={"X-API-Key": key})
|
||||
|
||||
assert response.status_code == 403
|
||||
@@ -232,8 +232,8 @@ class TestAuthMeAPI:
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_me_with_api_key_bearer(self, async_client: AsyncClient, db_session):
|
||||
"""Verify /me returns synthetic admin user when using API key via Bearer token."""
|
||||
async def test_me_with_ownerless_api_key_bearer(self, async_client: AsyncClient, db_session):
|
||||
"""A legacy key has no identity to report, but no longer claims admin (#1894)."""
|
||||
from backend.app.core.auth import generate_api_key
|
||||
from backend.app.models.api_key import APIKey
|
||||
|
||||
@@ -253,15 +253,18 @@ class TestAuthMeAPI:
|
||||
result = response.json()
|
||||
assert result["id"] == 0
|
||||
assert result["username"].startswith("api-key:")
|
||||
assert result["role"] == "admin"
|
||||
assert result["is_admin"] is True
|
||||
assert result["role"] != "admin"
|
||||
assert result["is_admin"] is False
|
||||
assert result["is_active"] is True
|
||||
# can_read_status defaults True, so the scope-derived set is non-empty
|
||||
# -- but it is a set, not "every permission there is".
|
||||
assert len(result["permissions"]) > 0
|
||||
assert "users:create" not in result["permissions"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_me_with_api_key_header(self, async_client: AsyncClient, db_session):
|
||||
"""Verify /me returns synthetic admin user when using X-API-Key header."""
|
||||
async def test_me_with_ownerless_api_key_header(self, async_client: AsyncClient, db_session):
|
||||
"""Same as above via the X-API-Key header rather than Bearer."""
|
||||
from backend.app.core.auth import generate_api_key
|
||||
from backend.app.models.api_key import APIKey
|
||||
|
||||
@@ -279,7 +282,7 @@ class TestAuthMeAPI:
|
||||
result = response.json()
|
||||
assert result["id"] == 0
|
||||
assert result["username"].startswith("api-key:")
|
||||
assert result["is_admin"] is True
|
||||
assert result["is_admin"] is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
@@ -292,6 +295,127 @@ class TestAuthMeAPI:
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
async def _owned_key(self, async_client: AsyncClient, db_session, **scopes):
|
||||
"""Set up auth and return (owner, full_key) for a key with ``scopes``.
|
||||
|
||||
The owner is given an email and a group explicitly rather than relying
|
||||
on what /auth/setup happens to seed, so the assertions about what /me
|
||||
withholds cannot pass vacuously.
|
||||
"""
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import selectinload
|
||||
|
||||
from backend.app.core.auth import generate_api_key
|
||||
from backend.app.models.api_key import APIKey
|
||||
from backend.app.models.group import Group
|
||||
from backend.app.models.user import User
|
||||
|
||||
await async_client.post(
|
||||
"/api/v1/auth/setup",
|
||||
json={
|
||||
"auth_enabled": True,
|
||||
"admin_username": "keyowner",
|
||||
"admin_password": "KeyPass1!",
|
||||
},
|
||||
)
|
||||
owner = (
|
||||
await db_session.execute(select(User).where(User.username == "keyowner").options(selectinload(User.groups)))
|
||||
).scalar_one()
|
||||
owner.email = "keyowner@example.invalid"
|
||||
group = Group(name="key-owner-group", description="t", permissions=["printers:read"], is_system=False)
|
||||
db_session.add(group)
|
||||
await db_session.flush()
|
||||
owner.groups.append(group)
|
||||
|
||||
full_key, key_hash, key_prefix = generate_api_key()
|
||||
db_session.add(
|
||||
APIKey(name="owned", key_hash=key_hash, key_prefix=key_prefix, enabled=True, user_id=owner.id, **scopes)
|
||||
)
|
||||
await db_session.commit()
|
||||
return owner, full_key
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_me_reports_the_key_owner_not_a_synthetic_admin(self, async_client: AsyncClient, db_session):
|
||||
"""The id is the point of #1894 -- it is what created_by_id filters on."""
|
||||
owner, full_key = await self._owned_key(async_client, db_session)
|
||||
|
||||
response = await async_client.get("/api/v1/auth/me", headers={"X-API-Key": full_key})
|
||||
|
||||
assert response.status_code == 200
|
||||
result = response.json()
|
||||
assert result["id"] == owner.id
|
||||
assert result["username"] == "keyowner"
|
||||
# The owner is an admin; the key still is not, because no key reaches
|
||||
# an administrative route regardless of who owns it.
|
||||
assert result["is_admin"] is False
|
||||
assert result["role"] != "admin"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_me_withholds_owner_email_and_groups(self, async_client: AsyncClient, db_session):
|
||||
"""Identity, not the owner's profile -- anyone holding the key sees this."""
|
||||
owner, full_key = await self._owned_key(async_client, db_session)
|
||||
assert owner.email is not None and owner.groups # the helper made both non-empty
|
||||
|
||||
result = (await async_client.get("/api/v1/auth/me", headers={"X-API-Key": full_key})).json()
|
||||
|
||||
assert result["email"] is None
|
||||
assert result["groups"] == []
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_me_permissions_track_the_key_scopes_not_the_owner(self, async_client: AsyncClient, db_session):
|
||||
"""A key owned by an admin still reports only what its flags allow."""
|
||||
_, full_key = await self._owned_key(
|
||||
async_client,
|
||||
db_session,
|
||||
can_read_status=True,
|
||||
can_control_printer=False,
|
||||
can_queue=False,
|
||||
)
|
||||
|
||||
perms = (await async_client.get("/api/v1/auth/me", headers={"X-API-Key": full_key})).json()["permissions"]
|
||||
|
||||
assert "printers:read" in perms # can_read_status
|
||||
assert "printers:control" not in perms # can_control_printer is off
|
||||
assert "queue:create" not in perms # can_queue is off
|
||||
assert "users:create" not in perms # administrative: unmapped for keys
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_me_permissions_are_exactly_what_the_gate_admits(self, async_client: AsyncClient, db_session):
|
||||
"""/me must not drift from _check_apikey_permissions.
|
||||
|
||||
The whole defect in #1894 was a /me response that described a different
|
||||
credential than the one the gate enforces, so pin them to each other
|
||||
rather than to a hand-written list that can rot. The owner is threaded
|
||||
through both sides for the same reason -- the gate narrows to the
|
||||
owner's permissions, so a check that skipped the owner would stop
|
||||
catching drift the moment the owner is not an administrator.
|
||||
"""
|
||||
from fastapi import HTTPException
|
||||
from sqlalchemy import select
|
||||
|
||||
from backend.app.core.auth import _check_apikey_permissions, resolve_apikey_owner
|
||||
from backend.app.core.permissions import ALL_PERMISSIONS
|
||||
from backend.app.models.api_key import APIKey
|
||||
|
||||
_, full_key = await self._owned_key(async_client, db_session, can_read_status=True, can_control_printer=False)
|
||||
|
||||
response = await async_client.get("/api/v1/auth/me", headers={"X-API-Key": full_key})
|
||||
reported = set(response.json()["permissions"])
|
||||
|
||||
key = (await db_session.execute(select(APIKey).where(APIKey.name == "owned"))).scalar_one()
|
||||
owner = await resolve_apikey_owner(db_session, key)
|
||||
for perm in ALL_PERMISSIONS:
|
||||
try:
|
||||
_check_apikey_permissions(key, [perm], owner=owner)
|
||||
except HTTPException:
|
||||
assert perm not in reported, f"/me reports '{perm}' but the gate denies it"
|
||||
else:
|
||||
assert perm in reported, f"the gate admits '{perm}' but /me omits it"
|
||||
|
||||
|
||||
class TestUsersAPI:
|
||||
"""Integration tests for /api/v1/users/ endpoints."""
|
||||
|
||||
@@ -30,6 +30,7 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
||||
from backend.app.core.auth import generate_api_key
|
||||
from backend.app.core.config import settings as app_settings
|
||||
from backend.app.models.api_key import APIKey
|
||||
from backend.app.models.group import Group
|
||||
from backend.app.models.print_queue import PrintQueueItem
|
||||
from backend.app.models.user import User
|
||||
|
||||
@@ -173,7 +174,14 @@ class TestWebhookQueueAddAttribution:
|
||||
async def test_credits_the_key_owner(self, async_client: AsyncClient, db_session, test_engine, printer_and_archive):
|
||||
printer, archive = printer_and_archive
|
||||
|
||||
owner = User(username="keyowner", password_hash="x", is_active=True)
|
||||
# The owner needs queue:create in their own right: a key is capped by
|
||||
# its owner's permissions (#1894), so a bare account with no groups
|
||||
# cannot queue through a key however its scope flags are set. This test
|
||||
# is about who the row is credited to, not about the gate.
|
||||
group = Group(name="queue-writers", description="t", permissions=["queue:create"], is_system=False)
|
||||
db_session.add(group)
|
||||
await db_session.flush()
|
||||
owner = User(username="keyowner", password_hash="x", is_active=True, groups=[group])
|
||||
db_session.add(owner)
|
||||
await db_session.commit()
|
||||
await db_session.refresh(owner)
|
||||
|
||||
@@ -0,0 +1,180 @@
|
||||
"""GET /api/v1/users/slim -- the id -> username mapping for API clients (#1894).
|
||||
|
||||
An API key could already read global archive stats and filter them by
|
||||
``created_by_id`` (for API-keyed requests the permission deps return None as
|
||||
``current_user``, so the ``stats:filter_by_user`` guard short-circuits), but
|
||||
had no way to discover which id belonged to whom: the full listing is gated on
|
||||
``users:read``, which is unmapped in the API-key scope allowlist and therefore
|
||||
administrative.
|
||||
|
||||
The slim listing closes that gap without handing keys the full user objects.
|
||||
These tests pin both halves: that it answers for a key, and that it stays
|
||||
narrow while the full listing stays admin-only.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
from sqlalchemy import select
|
||||
|
||||
from backend.app.core.auth import generate_api_key
|
||||
from backend.app.models.api_key import APIKey
|
||||
from backend.app.models.group import Group
|
||||
from backend.app.models.user import User
|
||||
|
||||
|
||||
async def _setup_and_login(async_client: AsyncClient) -> str:
|
||||
await async_client.post(
|
||||
"/api/v1/auth/setup",
|
||||
json={"auth_enabled": True, "admin_username": "slimadmin", "admin_password": "SlimPass1!"},
|
||||
)
|
||||
login = await async_client.post(
|
||||
"/api/v1/auth/login",
|
||||
json={"username": "slimadmin", "password": "SlimPass1!"},
|
||||
)
|
||||
return login.json()["access_token"]
|
||||
|
||||
|
||||
async def _add_key(db_session, *, user_id: int | None = None, **scopes) -> str:
|
||||
full_key, key_hash, key_prefix = generate_api_key()
|
||||
db_session.add(
|
||||
APIKey(name="slim-test", key_hash=key_hash, key_prefix=key_prefix, enabled=True, user_id=user_id, **scopes)
|
||||
)
|
||||
await db_session.commit()
|
||||
return full_key
|
||||
|
||||
|
||||
async def _add_user(db_session, username: str, **kwargs) -> User:
|
||||
from backend.app.core.auth import get_password_hash
|
||||
|
||||
user = User(
|
||||
username=username,
|
||||
password_hash=get_password_hash("Whatever1!"),
|
||||
email=f"{username}@example.invalid",
|
||||
role="user",
|
||||
is_active=True,
|
||||
**kwargs,
|
||||
)
|
||||
db_session.add(user)
|
||||
await db_session.commit()
|
||||
return user
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_slim_returns_only_id_and_username(async_client: AsyncClient, db_session):
|
||||
"""The response shape is the contract -- no emails, roles, or permissions."""
|
||||
token = await _setup_and_login(async_client)
|
||||
await _add_user(db_session, "bob")
|
||||
|
||||
response = await async_client.get("/api/v1/users/slim", headers={"Authorization": f"Bearer {token}"})
|
||||
|
||||
assert response.status_code == 200
|
||||
rows = response.json()
|
||||
assert rows, "expected at least the admin created by setup"
|
||||
for row in rows:
|
||||
assert set(row) == {"id", "username"}
|
||||
assert "bob" in [row["username"] for row in rows]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_slim_is_reachable_with_an_api_key(async_client: AsyncClient, db_session):
|
||||
"""The point of the issue: a key can resolve the ids it already filters on."""
|
||||
await _setup_and_login(async_client)
|
||||
owner = (await db_session.execute(select(User).where(User.username == "slimadmin"))).scalar_one()
|
||||
full_key = await _add_key(db_session, user_id=owner.id, can_read_status=True)
|
||||
|
||||
header = await async_client.get("/api/v1/users/slim", headers={"X-API-Key": full_key})
|
||||
bearer = await async_client.get("/api/v1/users/slim", headers={"Authorization": f"Bearer {full_key}"})
|
||||
|
||||
assert header.status_code == 200
|
||||
assert bearer.status_code == 200
|
||||
assert {row["username"] for row in header.json()} == {"slimadmin"}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_slim_needs_can_read_status(async_client: AsyncClient, db_session):
|
||||
"""A key without the read scope gets nothing, same as any other read route."""
|
||||
await _setup_and_login(async_client)
|
||||
owner = (await db_session.execute(select(User).where(User.username == "slimadmin"))).scalar_one()
|
||||
full_key = await _add_key(db_session, user_id=owner.id, can_read_status=False)
|
||||
|
||||
response = await async_client.get("/api/v1/users/slim", headers={"X-API-Key": full_key})
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_full_listing_stays_admin_only_for_api_keys(async_client: AsyncClient, db_session):
|
||||
"""Regression guard: widening the slim route must not widen the full one.
|
||||
|
||||
``users:read`` returns emails, group membership and the complete permission
|
||||
set for every account. It has to stay unmapped in the scope allowlist.
|
||||
"""
|
||||
await _setup_and_login(async_client)
|
||||
owner = (await db_session.execute(select(User).where(User.username == "slimadmin"))).scalar_one()
|
||||
full_key = await _add_key(db_session, user_id=owner.id, can_read_status=True)
|
||||
|
||||
response = await async_client.get("/api/v1/users", headers={"X-API-Key": full_key})
|
||||
|
||||
assert response.status_code == 403
|
||||
assert "administrative" in response.json()["detail"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_slim_is_not_parsed_as_a_user_id(async_client: AsyncClient, db_session):
|
||||
"""Route ordering. Declared after /{user_id}, "slim" would 422 as an int."""
|
||||
token = await _setup_and_login(async_client)
|
||||
|
||||
response = await async_client.get("/api/v1/users/slim", headers={"Authorization": f"Bearer {token}"})
|
||||
|
||||
assert response.status_code != 422
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_a_group_with_only_users_read_still_reaches_slim(async_client: AsyncClient, db_session):
|
||||
"""``users:read`` is strictly broader, so it must pass the any-of gate.
|
||||
|
||||
Without this, every existing custom group holding ``users:read`` would need
|
||||
a permission backfill before the frontend could ever move to this route.
|
||||
"""
|
||||
await _setup_and_login(async_client)
|
||||
group = Group(name="readers", description="t", permissions=["users:read"], is_system=False)
|
||||
db_session.add(group)
|
||||
await db_session.flush()
|
||||
await _add_user(db_session, "reader", groups=[group])
|
||||
|
||||
login = await async_client.post(
|
||||
"/api/v1/auth/login",
|
||||
json={"username": "reader", "password": "Whatever1!"},
|
||||
)
|
||||
token = login.json()["access_token"]
|
||||
|
||||
response = await async_client.get("/api/v1/users/slim", headers={"Authorization": f"Bearer {token}"})
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_a_group_with_only_slim_cannot_read_the_full_listing(async_client: AsyncClient, db_session):
|
||||
"""The narrow grant has to actually be narrower for JWT users too."""
|
||||
await _setup_and_login(async_client)
|
||||
group = Group(name="slim-only", description="t", permissions=["users:read_slim"], is_system=False)
|
||||
db_session.add(group)
|
||||
await db_session.flush()
|
||||
await _add_user(db_session, "slimonly", groups=[group])
|
||||
|
||||
login = await async_client.post(
|
||||
"/api/v1/auth/login",
|
||||
json={"username": "slimonly", "password": "Whatever1!"},
|
||||
)
|
||||
token = login.json()["access_token"]
|
||||
headers = {"Authorization": f"Bearer {token}"}
|
||||
|
||||
assert (await async_client.get("/api/v1/users/slim", headers=headers)).status_code == 200
|
||||
assert (await async_client.get("/api/v1/users", headers=headers)).status_code == 403
|
||||
Reference in New Issue
Block a user