diff --git a/paskia/db/operations.py b/paskia/db/operations.py index 72444a7..b476058 100644 --- a/paskia/db/operations.py +++ b/paskia/db/operations.py @@ -79,16 +79,18 @@ class DB: async def load(self, db_path: str | None = None) -> None: """Load data from JSONL change log. - If file doesn't exist, keeps the initialized empty structure and + If file doesn't exist or is empty, keeps the initialized empty structure and sets _previous_builtins to {} for creating a new database. """ if db_path is not None: self.db_path = Path(db_path) try: data_dict = await load_jsonl(self.db_path) - self._data = _json_decoder.decode(_json_encoder.encode(data_dict)) - # Track the JSONL file state directly - this is what we diff against - self._previous_builtins = data_dict + if data_dict: # Only decode if we have data + self._data = _json_decoder.decode(_json_encoder.encode(data_dict)) + # Track the JSONL file state directly - this is what we diff against + self._previous_builtins = data_dict + # If data_dict is empty, keep initialized _data and _previous_builtins = {} except ValueError: if self.db_path.exists(): raise # File exists but failed to load - re-raise diff --git a/paskia/db/structs.py b/paskia/db/structs.py index 6ccd0ce..630731a 100644 --- a/paskia/db/structs.py +++ b/paskia/db/structs.py @@ -100,7 +100,7 @@ class _OrgData(msgspec.Struct): class _RoleData(msgspec.Struct): org: UUID display_name: str - permissions: dict[UUID, bool] # permission_uuid -> True + permissions: dict[UUID, bool] = {} # permission_uuid -> True class _UserData(msgspec.Struct): diff --git a/paskia/fastapi/admin.py b/paskia/fastapi/admin.py index b74e3b1..4e23814 100644 --- a/paskia/fastapi/admin.py +++ b/paskia/fastapi/admin.py @@ -273,15 +273,23 @@ async def admin_create_role( perms = payload.get("permissions") or [] org = db.get_organization(str(org_uuid)) grantable = set(org.permissions or []) + + # Normalize permission IDs to UUIDs + permission_uuids = [] for pid in perms: - db.get_permission(pid) - if pid not in grantable: + perm = db.get_permission(pid) + if not perm: + raise ValueError(f"Permission {pid} not found") + perm_uuid_str = str(perm.uuid) + if perm_uuid_str not in grantable: raise ValueError(f"Permission not grantable by org: {pid}") + permission_uuids.append(perm_uuid_str) + role = RoleDC( uuid=role_uuid, org_uuid=org_uuid, display_name=display_name, - permissions=perms, + permissions=permission_uuids, ) db.create_role(role, ctx=ctx) return {"uuid": str(role_uuid)} @@ -506,10 +514,13 @@ async def admin_update_user_role( if ctx.user.uuid == user_uuid: new_role_obj = next((r for r in roles if r.display_name == new_role), None) if new_role_obj: # pragma: no branch - always true, role validated above - has_admin_access = ( - "auth:admin" in new_role_obj.permissions - or "auth:org:admin" in new_role_obj.permissions - ) + # Check if any permission in the new role is an admin permission + has_admin_access = False + for perm_uuid in new_role_obj.permissions: + perm = db.get_permission(perm_uuid) + if perm and perm.scope in ["auth:admin", "auth:org:admin"]: + has_admin_access = True + break if not has_admin_access: raise ValueError( "Cannot change your own role to one without admin permissions" @@ -917,9 +928,9 @@ async def admin_list_permissions(request: Request, auth=AUTH_COOKIE): if is_global_admin(ctx): return [_perm_to_dict(p) for p in perms] - # Org admins only see permissions their org can grant + # Org admins only see permissions their org can grant (by UUID) grantable = set(ctx.org.permissions or []) - filtered_perms = [p for p in perms if p.scope in grantable] + filtered_perms = [p for p in perms if str(p.uuid) in grantable] return [_perm_to_dict(p) for p in filtered_perms] diff --git a/tests/conftest.py b/tests/conftest.py index 745221e..0af7d11 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -87,7 +87,7 @@ async def test_org(test_db: DB, admin_permission: Permission) -> Org: org = Org( uuid=uuid7.create(), display_name="Test Organization", - permissions=["auth:admin"], # Org can grant this permission + permissions=[str(admin_permission.uuid)], # Org can grant this permission ) create_organization(org) return org @@ -131,7 +131,7 @@ async def test_role( uuid=uuid7.create(), org_uuid=test_org.uuid, display_name="Test Admin Role", - permissions=["auth:admin", "auth:org:admin"], + permissions=[str(admin_permission.uuid), str(org_admin_permission.uuid)], ) create_role(role) return role diff --git a/tests/test_admin.py b/tests/test_admin.py index bb4611b..f7ee8ad 100644 --- a/tests/test_admin.py +++ b/tests/test_admin.py @@ -61,7 +61,7 @@ async def second_org_role( uuid=uuid7.create(), org_uuid=second_org.uuid, display_name="Second Org Admin Role", - permissions=["auth:admin"], + permissions=[str(admin_permission.uuid)], ) create_role(role) return role @@ -120,13 +120,13 @@ async def second_org_session_token( @pytest_asyncio.fixture(scope="function") -async def org_admin_role(test_db: DB, test_org: Org, org_admin_permission) -> Role: +async def org_admin_role(test_db: DB, test_org: Org, org_admin_permission: Permission) -> Role: """Create a role with org admin permission only (no global admin).""" role = Role( uuid=uuid7.create(), org_uuid=test_org.uuid, display_name="Org Admin Role", - permissions=["auth:org:admin"], + permissions=[str(org_admin_permission.uuid)], ) create_role(role) return role @@ -666,7 +666,7 @@ class TestAdminRoles: ): """Admin should be able to add grantable permissions to role.""" response = await client.post( - f"/auth/api/admin/orgs/{test_org.uuid}/roles/{user_role.uuid}/permissions/{grantable_permission.scope}", + f"/auth/api/admin/orgs/{test_org.uuid}/roles/{user_role.uuid}/permissions/{grantable_permission.uuid}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 @@ -691,7 +691,7 @@ class TestAdminRoles: create_permission(perm) response = await client.post( - f"/auth/api/admin/orgs/{test_org.uuid}/roles/{user_role.uuid}/permissions/test:not:grantable:update", + f"/auth/api/admin/orgs/{test_org.uuid}/roles/{user_role.uuid}/permissions/{perm.uuid}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 400 @@ -700,20 +700,26 @@ class TestAdminRoles: @pytest.mark.asyncio async def test_update_own_role_cannot_remove_admin( - self, client: httpx.AsyncClient, session_token: str, test_org, test_role + self, + client: httpx.AsyncClient, + session_token: str, + test_org, + test_role, + admin_permission, + org_admin_permission, ): """Admin cannot remove their own admin permissions.""" # test_role has both auth:admin and auth:org:admin # Remove auth:admin first (should succeed since org:admin remains) response = await client.delete( - f"/auth/api/admin/orgs/{test_org.uuid}/roles/{test_role.uuid}/permissions/auth:admin", + f"/auth/api/admin/orgs/{test_org.uuid}/roles/{test_role.uuid}/permissions/{admin_permission.uuid}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 # Now try to remove auth:org:admin (should fail - would leave no admin access) response = await client.delete( - f"/auth/api/admin/orgs/{test_org.uuid}/roles/{test_role.uuid}/permissions/auth:org:admin", + f"/auth/api/admin/orgs/{test_org.uuid}/roles/{test_role.uuid}/permissions/{org_admin_permission.uuid}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 400 @@ -1264,14 +1270,14 @@ class TestAdminSessions: async def test_delete_session_invalid_id( self, client: httpx.AsyncClient, session_token: str, test_org, test_user ): - """Deleting session with invalid ID format should fail.""" + """Deleting session with invalid/non-existent ID should fail.""" response = await client.delete( f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}/sessions/invalid!!id", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) - assert response.status_code == 400 + assert response.status_code == 404 data = response.json() - assert "Invalid session identifier" in data["detail"] + assert "Session not found" in data["detail"] @pytest.mark.asyncio async def test_delete_session_not_found( diff --git a/tests/test_user.py b/tests/test_user.py index 8c2c1af..7350f12 100644 --- a/tests/test_user.py +++ b/tests/test_user.py @@ -117,7 +117,7 @@ class TestUserSessionManagement: "/auth/api/user/session/invalid-session-id", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) - assert response.status_code == 400 + assert response.status_code == 404 # Not found (no format validation) @pytest.mark.asyncio async def test_delete_nonexistent_session_returns_404(