Add required privileges to all endpoints

tests/controller/test_rbac.py

@@ -16,51 +16,118 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import pytest
+import pytest_asyncio
 
 from fastapi import FastAPI, status
 from httpx import AsyncClient
 
 from sqlalchemy.ext.asyncio import AsyncSession
 from gns3server.db.repositories.rbac import RbacRepository
+from gns3server.db.repositories.users import UsersRepository
+from gns3server.schemas.controller.rbac import ACECreate
 from gns3server.db.models import User
 
 pytestmark = pytest.mark.asyncio
 
-
-# class TestPermissions:
+# @pytest_asyncio.fixture
+# async def project_ace(db_session: AsyncSession):
 #
-#     @pytest.mark.parametrize(
-#         "method, path, result",
-#         (
-#             ("GET", "/users", False),
-#             ("GET", "/projects", True),
-#             ("GET", "/projects/e451ad73-2519-4f83-87fe-a8e821792d44", False),
-#             ("POST", "/projects", True),
-#             ("GET", "/templates", True),
-#             ("GET", "/templates/62e92cf1-244a-4486-8dae-b95439b54da9", False),
-#             ("POST", "/templates", True),
-#             ("GET", "/computes", True),
-#             ("GET", "/computes/local", True),
-#             ("GET", "/symbols", True),
-#             ("GET", "/symbols/default_symbols", True),
-#         ),
+#     group_id = (await UsersRepository(db_session).get_user_group_by_name("Users")).user_group_id
+#     role_id = (await RbacRepository(db_session).get_role_by_name("User")).role_id
+#     ace = ACECreate(
+#         path="/projects",
+#         ace_type="group",
+#         propagate=False,
+#         group_id=str(group_id),
+#         role_id=str(role_id)
 #     )
-#     async def test_default_permissions_user_group(
-#             self,
-#             app: FastAPI,
-#             authorized_client: AsyncClient,
-#             test_user: User,
-#             db_session: AsyncSession,
-#             method: str,
-#             path: str,
-#             result: bool
-#     ) -> None:
-#
-#         rbac_repo = RbacRepository(db_session)
-#         authorized = await rbac_repo.check_user_is_authorized(test_user.user_id, method, path)
-#         assert authorized == result
-#
-#
+#     await RbacRepository(db_session).create_ace(ace)
+
+
+class TestPrivileges:
+
+    @pytest.mark.parametrize(
+        "privilege, path, result",
+        (
+            ("User.Allocate", "/users", False),
+            ("Project.Allocate", "/projects", False),
+            ("Project.Allocate", "/projects", True),
+            ("Project.Audit", "/projects/e451ad73-2519-4f83-87fe-a8e821792d44", True),
+            ("Project.Audit", "/templates", False),
+            ("Template.Audit", "/templates", True),
+            ("Template.Allocate", "/templates", False),
+            ("Compute.Audit", "/computes", True),
+            ("Compute.Audit", "/computes/local", True),
+            ("Symbol.Audit", "/symbols", True),
+            ("Symbol.Audit", "/symbols/default_symbols", True),
+        ),
+    )
+    async def test_default_privileges_user_group(
+            self,
+            test_user: User,
+            db_session: AsyncSession,
+            privilege: str,
+            path: str,
+            result: bool
+    ) -> None:
+
+        # add an ACE for path
+        if result:
+            group_id = (await UsersRepository(db_session).get_user_group_by_name("Users")).user_group_id
+            role_id = (await RbacRepository(db_session).get_role_by_name("User")).role_id
+            ace = ACECreate(
+                path=path,
+                ace_type="group",
+                propagate=False,
+                group_id=str(group_id),
+                role_id=str(role_id)
+            )
+            await RbacRepository(db_session).create_ace(ace)
+
+        authorized = await RbacRepository(db_session).check_user_has_privilege(test_user.user_id, path, privilege)
+        assert authorized == result
+
+    async def test_propagate(self, test_user: User, db_session: AsyncSession):
+
+        privilege = "Project.Audit"
+        path = "/projects/44929147-47bb-460a-90ae-c782c4dbb6ef"
+        authorized = await RbacRepository(db_session).check_user_has_privilege(test_user.user_id, path, privilege)
+        assert authorized is False
+
+        ace = await RbacRepository(db_session).get_ace_by_path("/projects")
+        ace.propagate = True
+        await db_session.commit()
+
+        authorized = await RbacRepository(db_session).check_user_has_privilege(test_user.user_id, path, privilege)
+        assert authorized is True
+
+    async def test_allowed(self, test_user: User, db_session: AsyncSession):
+
+        ace = await RbacRepository(db_session).get_ace_by_path("/projects")
+        ace.allowed = False
+        ace.propagate = True
+        await db_session.commit()
+
+        privilege = "Project.Audit"
+        path = "/projects/44929147-47bb-460a-90ae-c782c4dbb6ef"
+        authorized = await RbacRepository(db_session).check_user_has_privilege(test_user.user_id, path, privilege)
+        assert authorized is False
+
+        # privileges on deeper levels replace those inherited from an upper level.
+        group_id = (await UsersRepository(db_session).get_user_group_by_name("Users")).user_group_id
+        role_id = (await RbacRepository(db_session).get_role_by_name("User")).role_id
+        ace = ACECreate(
+            path=path,
+            ace_type="group",
+            propagate=False,
+            group_id=str(group_id),
+            role_id=str(role_id)
+        )
+        await RbacRepository(db_session).create_ace(ace)
+
+        authorized = await RbacRepository(db_session).check_user_has_privilege(test_user.user_id, path, privilege)
+        assert authorized is True
+
 # class TestProjectsWithRbac:
 #
 #     async def test_admin_create_project(self, app: FastAPI, client: AsyncClient):
@@ -73,6 +140,7 @@ pytestmark = pytest.mark.asyncio
 #             self,
 #             app: FastAPI,
 #             authorized_client: AsyncClient,
+#             project_ace,
 #             test_user: User,
 #             db_session: AsyncSession
 #     ) -> None:
@@ -86,67 +154,67 @@ pytestmark = pytest.mark.asyncio
 #         permissions_in_db = await rbac_repo.get_user_permissions(test_user.user_id)
 #         assert len(permissions_in_db) == 1
 #         assert permissions_in_db[0].path == f"/projects/{project_id}/*"
-#
-#         response = await authorized_client.get(app.url_path_for("get_projects"))
-#         assert response.status_code == status.HTTP_200_OK
-#         projects = response.json()
-#         assert len(projects) == 1
-#
-#     async def test_admin_access_all_projects(self, app: FastAPI, client: AsyncClient):
-#
-#         response = await client.get(app.url_path_for("get_projects"))
-#         assert response.status_code == status.HTTP_200_OK
-#         projects = response.json()
-#         assert len(projects) == 2
-#
-#     async def test_admin_user_give_permission_on_project(
-#             self,
-#             app: FastAPI,
-#             client: AsyncClient,
-#             test_user: User
-#     ):
-#
-#         response = await client.get(app.url_path_for("get_projects"))
-#         assert response.status_code == status.HTTP_200_OK
-#         projects = response.json()
-#         project_id = None
-#         for project in projects:
-#             if project["name"] == "Admin project":
-#                 project_id = project["project_id"]
-#                 break
-#
-#         new_permission = {
-#             "methods": ["GET"],
-#             "path": f"/projects/{project_id}",
-#             "action": "ALLOW"
-#         }
-#         response = await client.post(app.url_path_for("create_permission"), json=new_permission)
-#         assert response.status_code == status.HTTP_201_CREATED
-#         permission_id = response.json()["permission_id"]
-#
-#         response = await client.put(
-#             app.url_path_for(
-#                 "add_permission_to_user",
-#                 user_id=test_user.user_id,
-#                 permission_id=permission_id
-#             )
-#         )
-#         assert response.status_code == status.HTTP_204_NO_CONTENT
-#
-#     async def test_user_access_admin_project(
-#             self,
-#             app: FastAPI,
-#             authorized_client: AsyncClient,
-#             test_user: User,
-#             db_session: AsyncSession
-#     ) -> None:
-#
-#         response = await authorized_client.get(app.url_path_for("get_projects"))
-#         assert response.status_code == status.HTTP_200_OK
-#         projects = response.json()
-#         assert len(projects) == 2
-#
-#
+        #
+        # response = await authorized_client.get(app.url_path_for("get_projects"))
+        # assert response.status_code == status.HTTP_200_OK
+        # projects = response.json()
+        # assert len(projects) == 1
+
+    # async def test_admin_access_all_projects(self, app: FastAPI, client: AsyncClient):
+    #
+    #     response = await client.get(app.url_path_for("get_projects"))
+    #     assert response.status_code == status.HTTP_200_OK
+    #     projects = response.json()
+    #     assert len(projects) == 2
+    #
+    # async def test_admin_user_give_permission_on_project(
+    #         self,
+    #         app: FastAPI,
+    #         client: AsyncClient,
+    #         test_user: User
+    # ):
+    #
+    #     response = await client.get(app.url_path_for("get_projects"))
+    #     assert response.status_code == status.HTTP_200_OK
+    #     projects = response.json()
+    #     project_id = None
+    #     for project in projects:
+    #         if project["name"] == "Admin project":
+    #             project_id = project["project_id"]
+    #             break
+    #
+    #     new_permission = {
+    #         "methods": ["GET"],
+    #         "path": f"/projects/{project_id}",
+    #         "action": "ALLOW"
+    #     }
+    #     response = await client.post(app.url_path_for("create_permission"), json=new_permission)
+    #     assert response.status_code == status.HTTP_201_CREATED
+    #     permission_id = response.json()["permission_id"]
+    #
+    #     response = await client.put(
+    #         app.url_path_for(
+    #             "add_permission_to_user",
+    #             user_id=test_user.user_id,
+    #             permission_id=permission_id
+    #         )
+    #     )
+    #     assert response.status_code == status.HTTP_204_NO_CONTENT
+    #
+    # async def test_user_access_admin_project(
+    #         self,
+    #         app: FastAPI,
+    #         authorized_client: AsyncClient,
+    #         test_user: User,
+    #         db_session: AsyncSession
+    # ) -> None:
+    #
+    #     response = await authorized_client.get(app.url_path_for("get_projects"))
+    #     assert response.status_code == status.HTTP_200_OK
+    #     projects = response.json()
+    #     assert len(projects) == 2
+    #
+
 # class TestTemplatesWithRbac:
 #
 #     async def test_admin_create_template(self, app: FastAPI, client: AsyncClient):