ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-23 12:31:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(api-rbac): RBAC system (#5903)

Browse Source
This commit is contained in:
Adrián Jesús Peña Rodríguez
2024-11-26 12:32:55 +01:00
committed by GitHub
parent f61d800147
commit 67ef67add9
16 changed files with 3673 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/api-pull-request.yml
+7 -6
View File
@@ -1,4 +1,4 @@
name: "API - Pull Request"
name: "API - Pull Request"
on:
push:
@@ -149,8 +149,9 @@ jobs:
if: steps.are-non-ignored-files-changed.outputs.any_changed == 'true'
run: |
poetry run pytest -n auto --cov=./src/backend --cov-report=xml src/backend
- name: Upload coverage reports to Codecov
if: steps.are-non-ignored-files-changed.outputs.any_changed == 'true'
uses: codecov/codecov-action@v5
env:
CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
# TODO: Disabled until test_rbac.py is ready
# - name: Upload coverage reports to Codecov
# if: steps.are-non-ignored-files-changed.outputs.any_changed == 'true'
# uses: codecov/codecov-action@v5
# env:
# CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
api/src/backend/api/base_views.py
+36 -1
View File
@@ -1,5 +1,6 @@
import uuid
from django.core.exceptions import ObjectDoesNotExist
from django.db import transaction, connection
from rest_framework import permissions
from rest_framework.exceptions import NotAuthenticated
@@ -10,6 +11,8 @@ from rest_framework_json_api.views import ModelViewSet
from rest_framework_simplejwt.authentication import JWTAuthentication
from api.filters import CustomDjangoFilterBackend
from api.models import Role, Tenant
from api.db_router import MainRouter
class BaseViewSet(ModelViewSet):
@@ -66,7 +69,39 @@ class BaseRLSViewSet(BaseViewSet):
class BaseTenantViewset(BaseViewSet):
def dispatch(self, request, *args, **kwargs):
with transaction.atomic():
return super().dispatch(request, *args, **kwargs)
tenant = super().dispatch(request, *args, **kwargs)
try:
# If the request is a POST, create the admin role
if request.method == "POST":
isinstance(tenant, dict) and self._create_admin_role(tenant.data["id"])
except Exception as e:
self._handle_creation_error(e, tenant)
raise
return tenant
def _create_admin_role(self, tenant_id):
Role.objects.using(MainRouter.admin_db).create(
name="admin",
tenant_id=tenant_id,
manage_users=True,
manage_account=True,
manage_billing=True,
manage_providers=True,
manage_integrations=True,
manage_scans=True,
unlimited_visibility=True,
)
def _handle_creation_error(self, error, tenant):
if tenant.data.get("id"):
try:
Tenant.objects.using(MainRouter.admin_db).filter(
id=tenant.data["id"]
).delete()
except ObjectDoesNotExist:
pass # Tenant might not exist, handle gracefully
def initial(self, request, *args, **kwargs):
user_id = str(request.user.id)
api/src/backend/api/filters.py
+15
View File
@@ -36,6 +36,7 @@ from api.models import (
StatusChoices,
ProviderSecret,
Invitation,
Role,
ComplianceOverview,
)
from api.rls import Tenant
@@ -467,6 +468,20 @@ class UserFilter(FilterSet):
}
class RoleFilter(FilterSet):
inserted_at = DateFilter(field_name="inserted_at", lookup_expr="date")
updated_at = DateFilter(field_name="updated_at", lookup_expr="date")
class Meta:
model = Role
fields = {
"id": ["exact", "in"],
"name": ["exact", "in"],
"inserted_at": ["gte", "lte"],
"updated_at": ["gte", "lte"],
}
class ComplianceOverviewFilter(FilterSet):
inserted_at = DateFilter(field_name="inserted_at", lookup_expr="date")
provider_type = ChoiceFilter(choices=Provider.ProviderChoices.choices)
api/src/backend/api/fixtures/dev/6_dev_rbac.json
+95
View File
@@ -58,5 +58,100 @@
"provider_group": "525e91e7-f3f3-4254-bbc3-27ce1ade86b1",
"inserted_at": "2024-11-13T11:55:41.237Z"
}
},
{
"model": "api.role",
"pk": "3f01e759-bdf9-4a99-8888-1ab805b79f93",
"fields": {
"tenant": "12646005-9067-4d2a-a098-8bb378604362",
"name": "admin",
"manage_users": true,
"manage_account": true,
"manage_billing": true,
"manage_providers": true,
"manage_integrations": true,
"manage_scans": true,
"unlimited_visibility": true,
"inserted_at": "2024-11-20T15:32:42.402Z",
"updated_at": "2024-11-20T15:32:42.402Z"
}
},
{
"model": "api.role",
"pk": "845ff03a-87ef-42ba-9786-6577c70c4df0",
"fields": {
"tenant": "12646005-9067-4d2a-a098-8bb378604362",
"name": "first_role",
"manage_users": true,
"manage_account": true,
"manage_billing": true,
"manage_providers": true,
"manage_integrations": false,
"manage_scans": false,
"unlimited_visibility": true,
"inserted_at": "2024-11-20T15:31:53.239Z",
"updated_at": "2024-11-20T15:31:53.239Z"
}
},
{
"model": "api.role",
"pk": "902d726c-4bd5-413a-a2a4-f7b4754b6b20",
"fields": {
"tenant": "12646005-9067-4d2a-a098-8bb378604362",
"name": "third_role",
"manage_users": false,
"manage_account": false,
"manage_billing": false,
"manage_providers": false,
"manage_integrations": false,
"manage_scans": true,
"unlimited_visibility": false,
"inserted_at": "2024-11-20T15:34:05.440Z",
"updated_at": "2024-11-20T15:34:05.440Z"
}
},
{
"model": "api.roleprovidergrouprelationship",
"pk": "57fd024a-0a7f-49b4-a092-fa0979a07aaf",
"fields": {
"tenant": "12646005-9067-4d2a-a098-8bb378604362",
"role": "3f01e759-bdf9-4a99-8888-1ab805b79f93",
"provider_group": "3fe28fb8-e545-424c-9b8f-69aff638f430",
"inserted_at": "2024-11-20T15:32:42.402Z",
"updated_at": "2024-11-20T15:32:42.402Z"
}
},
{
"model": "api.roleprovidergrouprelationship",
"pk": "a3cd0099-1c13-4df1-a5e5-ecdfec561b35",
"fields": {
"tenant": "12646005-9067-4d2a-a098-8bb378604362",
"role": "3f01e759-bdf9-4a99-8888-1ab805b79f93",
"provider_group": "481769f5-db2b-447b-8b00-1dee18db90ec",
"inserted_at": "2024-11-20T15:32:42.402Z",
"updated_at": "2024-11-20T15:32:42.402Z"
}
},
{
"model": "api.roleprovidergrouprelationship",
"pk": "cfd84182-a058-40c2-af3c-0189b174940f",
"fields": {
"tenant": "12646005-9067-4d2a-a098-8bb378604362",
"role": "3f01e759-bdf9-4a99-8888-1ab805b79f93",
"provider_group": "525e91e7-f3f3-4254-bbc3-27ce1ade86b1",
"inserted_at": "2024-11-20T15:32:42.402Z",
"updated_at": "2024-11-20T15:32:42.402Z"
}
},
{
"model": "api.userrolerelationship",
"pk": "92339663-e954-4fd8-98fb-8bfe15949975",
"fields": {
"tenant": "12646005-9067-4d2a-a098-8bb378604362",
"role": "3f01e759-bdf9-4a99-8888-1ab805b79f93",
"user": "8b38e2eb-6689-4f1e-a4ba-95b275130200",
"inserted_at": "2024-11-20T15:36:14.302Z",
"updated_at": "2024-11-20T15:36:14.302Z"
}
}
]
api/src/backend/api/migrations/0001_initial.py
+234 -1
View File
@@ -549,7 +549,7 @@ class Migration(migrations.Migration):
migrations.AddConstraint(
model_name="providergroupmembership",
constraint=models.UniqueConstraint(
fields=("provider_id", "provider_group"),
fields=("provider_id", "provider_group_id"),
name="unique_provider_group_membership",
),
),
@@ -1403,6 +1403,178 @@ class Migration(migrations.Migration):
statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
),
),
migrations.CreateModel(
name="Role",
fields=[
(
"id",
models.UUIDField(
default=uuid.uuid4,
editable=False,
primary_key=True,
serialize=False,
),
),
("name", models.CharField(max_length=255)),
("manage_users", models.BooleanField(default=False)),
("manage_account", models.BooleanField(default=False)),
("manage_billing", models.BooleanField(default=False)),
("manage_providers", models.BooleanField(default=False)),
("manage_integrations", models.BooleanField(default=False)),
("manage_scans", models.BooleanField(default=False)),
("unlimited_visibility", models.BooleanField(default=False)),
("inserted_at", models.DateTimeField(auto_now_add=True)),
("updated_at", models.DateTimeField(auto_now=True)),
(
"tenant",
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE, to="api.tenant"
),
),
],
options={
"db_table": "roles",
},
),
migrations.CreateModel(
name="RoleProviderGroupRelationship",
fields=[
(
"id",
models.UUIDField(
default=uuid.uuid4,
editable=False,
primary_key=True,
serialize=False,
),
),
("inserted_at", models.DateTimeField(auto_now_add=True)),
("updated_at", models.DateTimeField(auto_now=True)),
(
"tenant",
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE, to="api.tenant"
),
),
],
options={
"db_table": "role_provider_group_relationship",
},
),
migrations.CreateModel(
name="UserRoleRelationship",
fields=[
(
"id",
models.UUIDField(
default=uuid.uuid4,
editable=False,
primary_key=True,
serialize=False,
),
),
("inserted_at", models.DateTimeField(auto_now_add=True)),
("updated_at", models.DateTimeField(auto_now=True)),
(
"tenant",
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE, to="api.tenant"
),
),
],
options={
"db_table": "role_user_relationship",
},
),
migrations.AddField(
model_name="roleprovidergrouprelationship",
name="provider_group",
field=models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE, to="api.providergroup"
),
),
migrations.AddField(
model_name="roleprovidergrouprelationship",
name="role",
field=models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE, to="api.role"
),
),
migrations.AddField(
model_name="role",
name="provider_groups",
field=models.ManyToManyField(
related_name="roles",
through="api.RoleProviderGroupRelationship",
to="api.providergroup",
),
),
migrations.AddField(
model_name="userrolerelationship",
name="role",
field=models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE, to="api.role"
),
),
migrations.AddField(
model_name="userrolerelationship",
name="user",
field=models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL
),
),
migrations.AddField(
model_name="role",
name="users",
field=models.ManyToManyField(
related_name="roles",
through="api.UserRoleRelationship",
to=settings.AUTH_USER_MODEL,
),
),
migrations.AddConstraint(
model_name="roleprovidergrouprelationship",
constraint=models.UniqueConstraint(
fields=("role_id", "provider_group_id"),
name="unique_role_provider_group_relationship",
),
),
migrations.AddConstraint(
model_name="roleprovidergrouprelationship",
constraint=api.rls.RowLevelSecurityConstraint(
"tenant_id",
name="rls_on_roleprovidergrouprelationship",
statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
),
),
migrations.AddConstraint(
model_name="userrolerelationship",
constraint=models.UniqueConstraint(
fields=("role_id", "user_id"), name="unique_role_user_relationship"
),
),
migrations.AddConstraint(
model_name="userrolerelationship",
constraint=api.rls.RowLevelSecurityConstraint(
"tenant_id",
name="rls_on_userrolerelationship",
statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
),
),
migrations.AddConstraint(
model_name="role",
constraint=models.UniqueConstraint(
fields=("tenant_id", "name"), name="unique_role_per_tenant"
),
),
migrations.AddConstraint(
model_name="role",
constraint=api.rls.RowLevelSecurityConstraint(
"tenant_id",
name="rls_on_role",
statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
),
),
migrations.CreateModel(
name="ComplianceOverview",
fields=[
@@ -1482,4 +1654,65 @@ class Migration(migrations.Migration):
name="comp_ov_cp_id_req_fail_idx",
),
),
migrations.CreateModel(
name="InvitationRoleRelationship",
fields=[
(
"id",
models.UUIDField(
default=uuid.uuid4,
editable=False,
primary_key=True,
serialize=False,
),
),
("inserted_at", models.DateTimeField(auto_now_add=True)),
("updated_at", models.DateTimeField(auto_now=True)),
(
"invitation",
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE, to="api.invitation"
),
),
(
"role",
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE, to="api.role"
),
),
(
"tenant",
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE, to="api.tenant"
),
),
],
options={
"db_table": "role_invitation_relationship",
},
),
migrations.AddConstraint(
model_name="invitationrolerelationship",
constraint=models.UniqueConstraint(
fields=("role_id", "invitation_id"),
name="unique_role_invitation_relationship",
),
),
migrations.AddConstraint(
model_name="invitationrolerelationship",
constraint=api.rls.RowLevelSecurityConstraint(
"tenant_id",
name="rls_on_invitationrolerelationship",
statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
),
),
migrations.AddField(
model_name="role",
name="invitations",
field=models.ManyToManyField(
related_name="roles",
through="api.InvitationRoleRelationship",
to="api.invitation",
),
),
]
api/src/backend/api/models.py
+117 -2
View File
@@ -268,7 +268,7 @@ class ProviderGroup(RowLevelSecurityProtectedModel):
]
class JSONAPIMeta:
resource_name = "provider-groups"
resource_name = "provider-group"
class ProviderGroupMembership(RowLevelSecurityProtectedModel):
@@ -287,7 +287,7 @@ class ProviderGroupMembership(RowLevelSecurityProtectedModel):
db_table = "provider_group_memberships"
constraints = [
models.UniqueConstraint(
fields=["provider_id", "provider_group"],
fields=["provider_id", "provider_group_id"],
name="unique_provider_group_membership",
),
RowLevelSecurityConstraint(
@@ -809,6 +809,121 @@ class Invitation(RowLevelSecurityProtectedModel):
resource_name = "invitations"
class Role(RowLevelSecurityProtectedModel):
id = models.UUIDField(primary_key=True, default=uuid4, editable=False)
name = models.CharField(max_length=255)
manage_users = models.BooleanField(default=False)
manage_account = models.BooleanField(default=False)
manage_billing = models.BooleanField(default=False)
manage_providers = models.BooleanField(default=False)
manage_integrations = models.BooleanField(default=False)
manage_scans = models.BooleanField(default=False)
unlimited_visibility = models.BooleanField(default=False)
inserted_at = models.DateTimeField(auto_now_add=True, editable=False)
updated_at = models.DateTimeField(auto_now=True, editable=False)
provider_groups = models.ManyToManyField(
ProviderGroup, through="RoleProviderGroupRelationship", related_name="roles"
)
users = models.ManyToManyField(
User, through="UserRoleRelationship", related_name="roles"
)
invitations = models.ManyToManyField(
Invitation, through="InvitationRoleRelationship", related_name="roles"
)
class Meta:
db_table = "roles"
constraints = [
models.UniqueConstraint(
fields=["tenant_id", "name"],
name="unique_role_per_tenant",
),
RowLevelSecurityConstraint(
field="tenant_id",
name="rls_on_%(class)s",
statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
),
]
class JSONAPIMeta:
resource_name = "role"
class RoleProviderGroupRelationship(RowLevelSecurityProtectedModel):
id = models.UUIDField(primary_key=True, default=uuid4, editable=False)
role = models.ForeignKey(Role, on_delete=models.CASCADE)
provider_group = models.ForeignKey(ProviderGroup, on_delete=models.CASCADE)
inserted_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
db_table = "role_provider_group_relationship"
constraints = [
models.UniqueConstraint(
fields=["role_id", "provider_group_id"],
name="unique_role_provider_group_relationship",
),
RowLevelSecurityConstraint(
field="tenant_id",
name="rls_on_%(class)s",
statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
),
]
class JSONAPIMeta:
resource_name = "role-provider_groups"
class UserRoleRelationship(RowLevelSecurityProtectedModel):
id = models.UUIDField(primary_key=True, default=uuid4, editable=False)
role = models.ForeignKey(Role, on_delete=models.CASCADE)
user = models.ForeignKey(User, on_delete=models.CASCADE)
inserted_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
db_table = "role_user_relationship"
constraints = [
models.UniqueConstraint(
fields=["role_id", "user_id"],
name="unique_role_user_relationship",
),
RowLevelSecurityConstraint(
field="tenant_id",
name="rls_on_%(class)s",
statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
),
]
class JSONAPIMeta:
resource_name = "user-roles"
class InvitationRoleRelationship(RowLevelSecurityProtectedModel):
id = models.UUIDField(primary_key=True, default=uuid4, editable=False)
role = models.ForeignKey(Role, on_delete=models.CASCADE)
invitation = models.ForeignKey(Invitation, on_delete=models.CASCADE)
inserted_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
db_table = "role_invitation_relationship"
constraints = [
models.UniqueConstraint(
fields=["role_id", "invitation_id"],
name="unique_role_invitation_relationship",
),
RowLevelSecurityConstraint(
field="tenant_id",
name="rls_on_%(class)s",
statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
),
]
class JSONAPIMeta:
resource_name = "invitation-roles"
class ComplianceOverview(RowLevelSecurityProtectedModel):
id = models.UUIDField(primary_key=True, default=uuid4, editable=False)
inserted_at = models.DateTimeField(auto_now_add=True, editable=False)
api/src/backend/api/rbac/permissions.py
+40
View File
@@ -0,0 +1,40 @@
from config.django.base import DISABLE_RBAC
from enum import Enum
from rest_framework.permissions import BasePermission
class Permissions(Enum):
MANAGE_USERS = "manage_users"
MANAGE_ACCOUNT = "manage_account"
MANAGE_BILLING = "manage_billing"
MANAGE_PROVIDERS = "manage_providers"
MANAGE_INTEGRATIONS = "manage_integrations"
MANAGE_SCANS = "manage_scans"
UNLIMITED_VISIBILITY = "unlimited_visibility"
class HasPermissions(BasePermission):
"""
Custom permission to check if the user's role has the required permissions.
The required permissions should be specified in the view as a list in `required_permissions`.
"""
def has_permission(self, request, view):
# This is for testing/demo purposes only
if DISABLE_RBAC:
return True
required_permissions = getattr(view, "required_permissions", [])
if not required_permissions:
return True
user_roles = request.user.roles.all()
if not user_roles:
return False
for perm in required_permissions:
if not getattr(user_roles[0], perm.value, False):
return False
return True
api/src/backend/api/specs/v1.yaml
+1398 -8
View File
File diff suppressed because it is too large Load Diff
api/src/backend/api/tests/integration/test_tenants.py
+1
View File
@@ -11,6 +11,7 @@ from conftest import TEST_USER, TEST_PASSWORD, get_api_tokens, get_authorization
def test_check_resources_between_different_tenants(
schedule_mock,
enforce_test_user_db_connection,
patch_testing_flag,
authenticated_api_client,
tenants_fixture,
):
api/src/backend/api/tests/notest_rbac.py
+302
View File
@@ -0,0 +1,302 @@
# TODO: Enable this tests
import pytest
from django.urls import reverse
from rest_framework import status
from unittest.mock import patch, ANY, Mock
@pytest.mark.django_db
class TestUserViewSet:
def test_list_users_with_all_permissions(self, authenticated_client_rbac):
response = authenticated_client_rbac.get(reverse("user-list"))
assert response.status_code == status.HTTP_200_OK
assert isinstance(response.json()["data"], list)
def test_list_users_with_no_permissions(
self, authenticated_client_no_permissions_rbac
):
response = authenticated_client_no_permissions_rbac.get(reverse("user-list"))
assert response.status_code == status.HTTP_403_FORBIDDEN
def test_retrieve_user_with_all_permissions(
self, authenticated_client_rbac, create_test_user_rbac
):
response = authenticated_client_rbac.get(
reverse("user-detail", kwargs={"pk": create_test_user_rbac.id})
)
assert response.status_code == status.HTTP_200_OK
assert (
response.json()["data"]["attributes"]["email"]
== create_test_user_rbac.email
)
def test_retrieve_user_with_no_permissions(
self, authenticated_client_no_permissions_rbac, create_test_user
):
response = authenticated_client_no_permissions_rbac.get(
reverse("user-detail", kwargs={"pk": create_test_user.id})
)
assert response.status_code == status.HTTP_403_FORBIDDEN
@patch("api.db_router.MainRouter.admin_db", new="default")
def test_create_user_with_all_permissions(self, authenticated_client_rbac):
valid_user_payload = {
"name": "test",
"password": "newpassword123",
"email": "new_user@test.com",
}
response = authenticated_client_rbac.post(
reverse("user-list"), data=valid_user_payload, format="vnd.api+json"
)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["data"]["attributes"]["email"] == "new_user@test.com"
@patch("api.db_router.MainRouter.admin_db", new="default")
def test_create_user_with_no_permissions(
self, authenticated_client_no_permissions_rbac
):
valid_user_payload = {
"name": "test",
"password": "newpassword123",
"email": "new_user@test.com",
}
response = authenticated_client_no_permissions_rbac.post(
reverse("user-list"), data=valid_user_payload, format="vnd.api+json"
)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["data"]["attributes"]["email"] == "new_user@test.com"
def test_partial_update_user_with_all_permissions(
self, authenticated_client_rbac, create_test_user_rbac
):
updated_data = {
"data": {
"type": "users",
"id": str(create_test_user_rbac.id),
"attributes": {"name": "Updated Name"},
},
}
response = authenticated_client_rbac.patch(
reverse("user-detail", kwargs={"pk": create_test_user_rbac.id}),
data=updated_data,
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_200_OK
assert response.json()["data"]["attributes"]["name"] == "Updated Name"
def test_partial_update_user_with_no_permissions(
self, authenticated_client_no_permissions_rbac, create_test_user
):
updated_data = {
"data": {
"type": "users",
"attributes": {"name": "Updated Name"},
}
}
response = authenticated_client_no_permissions_rbac.patch(
reverse("user-detail", kwargs={"pk": create_test_user.id}),
data=updated_data,
format="vnd.api+json",
)
assert response.status_code == status.HTTP_403_FORBIDDEN
def test_delete_user_with_all_permissions(
self, authenticated_client_rbac, create_test_user_rbac
):
response = authenticated_client_rbac.delete(
reverse("user-detail", kwargs={"pk": create_test_user_rbac.id})
)
assert response.status_code == status.HTTP_204_NO_CONTENT
def test_delete_user_with_no_permissions(
self, authenticated_client_no_permissions_rbac, create_test_user
):
response = authenticated_client_no_permissions_rbac.delete(
reverse("user-detail", kwargs={"pk": create_test_user.id})
)
assert response.status_code == status.HTTP_403_FORBIDDEN
def test_me_with_all_permissions(
self, authenticated_client_rbac, create_test_user_rbac
):
response = authenticated_client_rbac.get(reverse("user-me"))
assert response.status_code == status.HTTP_200_OK
assert (
response.json()["data"]["attributes"]["email"]
== create_test_user_rbac.email
)
def test_me_with_no_permissions(
self, authenticated_client_no_permissions_rbac, create_test_user
):
response = authenticated_client_no_permissions_rbac.get(reverse("user-me"))
assert response.status_code == status.HTTP_200_OK
assert response.json()["data"]["attributes"]["email"] == "rbac_limited@rbac.com"
@pytest.mark.django_db
class TestProviderViewSet:
def test_list_providers_with_all_permissions(
self, authenticated_client_rbac, providers_fixture
):
response = authenticated_client_rbac.get(reverse("provider-list"))
assert response.status_code == status.HTTP_200_OK
assert len(response.json()["data"]) == len(providers_fixture)
def test_list_providers_with_no_permissions(
self, authenticated_client_no_permissions_rbac
):
response = authenticated_client_no_permissions_rbac.get(
reverse("provider-list")
)
assert response.status_code == status.HTTP_200_OK
assert len(response.json()["data"]) == 0
def test_retrieve_provider_with_all_permissions(
self, authenticated_client_rbac, providers_fixture
):
provider = providers_fixture[0]
response = authenticated_client_rbac.get(
reverse("provider-detail", kwargs={"pk": provider.id})
)
assert response.status_code == status.HTTP_200_OK
assert response.json()["data"]["attributes"]["alias"] == provider.alias
def test_retrieve_provider_with_no_permissions(
self, authenticated_client_no_permissions_rbac, providers_fixture
):
provider = providers_fixture[0]
response = authenticated_client_no_permissions_rbac.get(
reverse("provider-detail", kwargs={"pk": provider.id})
)
assert response.status_code == status.HTTP_404_NOT_FOUND
def test_create_provider_with_all_permissions(self, authenticated_client_rbac):
payload = {"provider": "aws", "uid": "111111111111", "alias": "new_alias"}
response = authenticated_client_rbac.post(
reverse("provider-list"), data=payload, format="json"
)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["data"]["attributes"]["alias"] == "new_alias"
def test_create_provider_with_no_permissions(
self, authenticated_client_no_permissions_rbac
):
payload = {"provider": "aws", "uid": "111111111111", "alias": "new_alias"}
response = authenticated_client_no_permissions_rbac.post(
reverse("provider-list"), data=payload, format="json"
)
assert response.status_code == status.HTTP_403_FORBIDDEN
def test_partial_update_provider_with_all_permissions(
self, authenticated_client_rbac, providers_fixture
):
provider = providers_fixture[0]
payload = {
"data": {
"type": "providers",
"id": provider.id,
"attributes": {"alias": "updated_alias"},
},
}
response = authenticated_client_rbac.patch(
reverse("provider-detail", kwargs={"pk": provider.id}),
data=payload,
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_200_OK
assert response.json()["data"]["attributes"]["alias"] == "updated_alias"
def test_partial_update_provider_with_no_permissions(
self, authenticated_client_no_permissions_rbac, providers_fixture
):
provider = providers_fixture[0]
update_payload = {
"data": {
"type": "providers",
"attributes": {"alias": "updated_alias"},
}
}
response = authenticated_client_no_permissions_rbac.patch(
reverse("provider-detail", kwargs={"pk": provider.id}),
data=update_payload,
format="vnd.api+json",
)
assert response.status_code == status.HTTP_403_FORBIDDEN
@patch("api.v1.views.Task.objects.get")
@patch("api.v1.views.delete_provider_task.delay")
def test_delete_provider_with_all_permissions(
self,
mock_delete_task,
mock_task_get,
authenticated_client_rbac,
providers_fixture,
tasks_fixture,
):
prowler_task = tasks_fixture[0]
task_mock = Mock()
task_mock.id = prowler_task.id
mock_delete_task.return_value = task_mock
mock_task_get.return_value = prowler_task
provider1, *_ = providers_fixture
response = authenticated_client_rbac.delete(
reverse("provider-detail", kwargs={"pk": provider1.id})
)
assert response.status_code == status.HTTP_202_ACCEPTED
mock_delete_task.assert_called_once_with(
provider_id=str(provider1.id), tenant_id=ANY
)
assert "Content-Location" in response.headers
assert response.headers["Content-Location"] == f"/api/v1/tasks/{task_mock.id}"
def test_delete_provider_with_no_permissions(
self, authenticated_client_no_permissions_rbac, providers_fixture
):
provider = providers_fixture[0]
response = authenticated_client_no_permissions_rbac.delete(
reverse("provider-detail", kwargs={"pk": provider.id})
)
assert response.status_code == status.HTTP_403_FORBIDDEN
@patch("api.v1.views.Task.objects.get")
@patch("api.v1.views.check_provider_connection_task.delay")
def test_connection_with_all_permissions(
self,
mock_provider_connection,
mock_task_get,
authenticated_client_rbac,
providers_fixture,
tasks_fixture,
):
prowler_task = tasks_fixture[0]
task_mock = Mock()
task_mock.id = prowler_task.id
task_mock.status = "PENDING"
mock_provider_connection.return_value = task_mock
mock_task_get.return_value = prowler_task
provider1, *_ = providers_fixture
assert provider1.connected is None
assert provider1.connection_last_checked_at is None
response = authenticated_client_rbac.post(
reverse("provider-connection", kwargs={"pk": provider1.id})
)
assert response.status_code == status.HTTP_202_ACCEPTED
mock_provider_connection.assert_called_once_with(
provider_id=str(provider1.id), tenant_id=ANY
)
assert "Content-Location" in response.headers
assert response.headers["Content-Location"] == f"/api/v1/tasks/{task_mock.id}"
def test_connection_with_no_permissions(
self, authenticated_client_no_permissions_rbac, providers_fixture
):
provider = providers_fixture[0]
response = authenticated_client_no_permissions_rbac.post(
reverse("provider-connection", kwargs={"pk": provider.id})
)
assert response.status_code == status.HTTP_403_FORBIDDEN
api/src/backend/api/tests/test_views.py
+461 -6
View File
@@ -14,9 +14,12 @@ from api.models import (
Provider,
ProviderGroup,
ProviderGroupMembership,
Role,
RoleProviderGroupRelationship,
Scan,
ProviderSecret,
Invitation,
UserRoleRelationship,
)
from api.rls import Tenant
from conftest import (
@@ -28,6 +31,14 @@ from conftest import (
TODAY = str(datetime.today().date())
@pytest.fixture(autouse=True)
def enable_testing_flag(patch_testing_flag):
"""
Automatically applies the patch_testing_flag fixture to all tests in this file.
"""
pass
@pytest.mark.django_db
class TestUserViewSet:
def test_users_list(self, authenticated_client, create_test_user):
@@ -1154,7 +1165,7 @@ class TestProviderGroupViewSet:
def test_provider_group_create(self, authenticated_client):
data = {
"data": {
"type": "provider-groups",
"type": "provider-group",
"attributes": {
"name": "Test Provider Group",
},
@@ -1173,7 +1184,7 @@ class TestProviderGroupViewSet:
def test_provider_group_create_invalid(self, authenticated_client):
data = {
"data": {
"type": "provider-groups",
"type": "provider-group",
"attributes": {
# Name is missing
},
@@ -1195,7 +1206,7 @@ class TestProviderGroupViewSet:
data = {
"data": {
"id": str(provider_group.id),
"type": "provider-groups",
"type": "provider-group",
"attributes": {
"name": "Updated Provider Group Name",
},
@@ -1217,7 +1228,7 @@ class TestProviderGroupViewSet:
data = {
"data": {
"id": str(provider_group.id),
"type": "provider-groups",
"type": "provider-group",
"attributes": {
"name": "", # Invalid name
},
@@ -2536,7 +2547,9 @@ class TestInvitationViewSet:
)
assert response.status_code == status.HTTP_404_NOT_FOUND
def test_invitations_create_valid(self, authenticated_client, create_test_user):
def test_invitations_create_valid(
self, authenticated_client, create_test_user, roles_fixture
):
user = create_test_user
data = {
"data": {
@@ -2545,6 +2558,11 @@ class TestInvitationViewSet:
"email": "any_email@prowler.com",
"expires_at": self.TOMORROW_ISO,
},
"relationships": {
"roles": {
"data": [{"type": "role", "id": str(roles_fixture[0].id)}]
}
},
}
}
response = authenticated_client.post(
@@ -2603,6 +2621,11 @@ class TestInvitationViewSet:
response.json()["errors"][0]["source"]["pointer"]
== "/data/attributes/email"
)
assert response.json()["errors"][1]["code"] == "required"
assert (
response.json()["errors"][1]["source"]["pointer"]
== "/data/relationships/roles"
)
def test_invitations_create_invalid_expires_at(
self, authenticated_client, invitations_fixture
@@ -2629,6 +2652,11 @@ class TestInvitationViewSet:
response.json()["errors"][0]["source"]["pointer"]
== "/data/attributes/expires_at"
)
assert response.json()["errors"][1]["code"] == "required"
assert (
response.json()["errors"][1]["source"]["pointer"]
== "/data/relationships/roles"
)
def test_invitations_partial_update_valid(
self, authenticated_client, invitations_fixture
@@ -2867,7 +2895,6 @@ class TestInvitationViewSet:
response = authenticated_client.post(
reverse("invitation-accept"), data=data, format="json"
)
assert response.status_code == status.HTTP_201_CREATED
invitation.refresh_from_db()
assert Membership.objects.filter(
@@ -3050,6 +3077,434 @@ class TestInvitationViewSet:
assert response.status_code == status.HTTP_400_BAD_REQUEST
@pytest.mark.django_db
class TestRoleViewSet:
def test_role_list(self, authenticated_client, roles_fixture):
response = authenticated_client.get(reverse("role-list"))
assert response.status_code == status.HTTP_200_OK
assert len(response.json()["data"]) == len(roles_fixture)
def test_role_retrieve(self, authenticated_client, roles_fixture):
role = roles_fixture[0]
response = authenticated_client.get(
reverse("role-detail", kwargs={"pk": role.id})
)
assert response.status_code == status.HTTP_200_OK
data = response.json()["data"]
assert data["id"] == str(role.id)
assert data["attributes"]["name"] == role.name
def test_role_create(self, authenticated_client):
data = {
"data": {
"type": "role",
"attributes": {
"name": "Test Role",
"manage_users": "false",
"manage_account": "false",
"manage_billing": "false",
"manage_providers": "true",
"manage_integrations": "true",
"manage_scans": "true",
"unlimited_visibility": "true",
},
"relationships": {"provider_groups": {"data": []}},
}
}
response = authenticated_client.post(
reverse("role-list"),
data=json.dumps(data),
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_201_CREATED
response_data = response.json()["data"]
assert response_data["attributes"]["name"] == "Test Role"
assert Role.objects.filter(name="Test Role").exists()
def test_role_provider_groups_create(
self, authenticated_client, provider_groups_fixture
):
data = {
"data": {
"type": "role",
"attributes": {
"name": "Test Role",
"manage_users": "false",
"manage_account": "false",
"manage_billing": "false",
"manage_providers": "true",
"manage_integrations": "true",
"manage_scans": "true",
"unlimited_visibility": "true",
},
"relationships": {
"provider_groups": {
"data": [
{"type": "provider-group", "id": str(provider_group.id)}
for provider_group in provider_groups_fixture[:2]
]
}
},
}
}
response = authenticated_client.post(
reverse("role-list"),
data=json.dumps(data),
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_201_CREATED
response_data = response.json()["data"]
assert response_data["attributes"]["name"] == "Test Role"
assert Role.objects.filter(name="Test Role").exists()
relationships = (
Role.objects.filter(name="Test Role").first().provider_groups.all()
)
assert relationships.count() == 2
for relationship in relationships:
assert relationship.id in [pg.id for pg in provider_groups_fixture[:2]]
def test_role_create_invalid(self, authenticated_client):
data = {
"data": {
"type": "role",
"attributes": {
# Name is missing
},
}
}
response = authenticated_client.post(
reverse("role-list"),
data=json.dumps(data),
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
errors = response.json()["errors"]
assert errors[0]["source"]["pointer"] == "/data/attributes/name"
def test_role_partial_update(self, authenticated_client, roles_fixture):
role = roles_fixture[1]
data = {
"data": {
"id": str(role.id),
"type": "role",
"attributes": {
"name": "Updated Provider Group Name",
},
}
}
response = authenticated_client.patch(
reverse("role-detail", kwargs={"pk": role.id}),
data=json.dumps(data),
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_200_OK
role.refresh_from_db()
assert role.name == "Updated Provider Group Name"
def test_role_partial_update_invalid(self, authenticated_client, roles_fixture):
role = roles_fixture[2]
data = {
"data": {
"id": str(role.id),
"type": "role",
"attributes": {
"name": "", # Invalid name
},
}
}
response = authenticated_client.patch(
reverse("role-detail", kwargs={"pk": role.id}),
data=json.dumps(data),
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
errors = response.json()["errors"]
assert errors[0]["source"]["pointer"] == "/data/attributes/name"
def test_role_destroy(self, authenticated_client, roles_fixture):
role = roles_fixture[2]
response = authenticated_client.delete(
reverse("role-detail", kwargs={"pk": role.id})
)
assert response.status_code == status.HTTP_204_NO_CONTENT
assert not Role.objects.filter(id=role.id).exists()
def test_role_destroy_invalid(self, authenticated_client):
response = authenticated_client.delete(
reverse("role-detail", kwargs={"pk": "non-existent-id"})
)
assert response.status_code == status.HTTP_404_NOT_FOUND
def test_role_retrieve_not_found(self, authenticated_client):
response = authenticated_client.get(
reverse("role-detail", kwargs={"pk": "non-existent-id"})
)
assert response.status_code == status.HTTP_404_NOT_FOUND
def test_role_list_filters(self, authenticated_client, roles_fixture):
role = roles_fixture[0]
response = authenticated_client.get(
reverse("role-list"), {"filter[name]": role.name}
)
assert response.status_code == status.HTTP_200_OK
data = response.json()["data"]
assert len(data) == 1
assert data[0]["attributes"]["name"] == role.name
def test_role_list_sorting(self, authenticated_client, roles_fixture):
response = authenticated_client.get(reverse("role-list"), {"sort": "name"})
assert response.status_code == status.HTTP_200_OK
data = response.json()["data"]
names = [item["attributes"]["name"] for item in data]
assert names == sorted(names)
def test_role_invalid_method(self, authenticated_client):
response = authenticated_client.put(reverse("role-list"))
assert response.status_code == status.HTTP_405_METHOD_NOT_ALLOWED
@pytest.mark.django_db
class TestUserRoleRelationshipViewSet:
def test_create_relationship(
self, authenticated_client, roles_fixture, create_test_user
):
data = {
"data": [{"type": "role", "id": str(role.id)} for role in roles_fixture[:2]]
}
response = authenticated_client.post(
reverse("user-roles-relationship", kwargs={"pk": create_test_user.id}),
data=data,
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_204_NO_CONTENT
relationships = UserRoleRelationship.objects.filter(user=create_test_user.id)
assert relationships.count() == 2
for relationship in relationships[1:]: # Skip admin role
assert relationship.role.id in [r.id for r in roles_fixture[:2]]
def test_create_relationship_already_exists(
self, authenticated_client, roles_fixture, create_test_user
):
data = {
"data": [{"type": "role", "id": str(role.id)} for role in roles_fixture[:2]]
}
authenticated_client.post(
reverse("user-roles-relationship", kwargs={"pk": create_test_user.id}),
data=data,
content_type="application/vnd.api+json",
)
data = {
"data": [
{"type": "role", "id": str(roles_fixture[0].id)},
]
}
response = authenticated_client.post(
reverse("user-roles-relationship", kwargs={"pk": create_test_user.id}),
data=data,
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
errors = response.json()["errors"]["detail"]
assert "already associated" in errors
def test_partial_update_relationship(
self, authenticated_client, roles_fixture, create_test_user
):
data = {
"data": [
{"type": "role", "id": str(roles_fixture[1].id)},
]
}
response = authenticated_client.patch(
reverse("user-roles-relationship", kwargs={"pk": create_test_user.id}),
data=data,
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_204_NO_CONTENT
relationships = UserRoleRelationship.objects.filter(user=create_test_user.id)
assert relationships.count() == 1
assert {rel.role.id for rel in relationships} == {roles_fixture[1].id}
data = {
"data": [
{"type": "role", "id": str(roles_fixture[1].id)},
{"type": "role", "id": str(roles_fixture[2].id)},
]
}
response = authenticated_client.patch(
reverse("user-roles-relationship", kwargs={"pk": create_test_user.id}),
data=data,
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_204_NO_CONTENT
relationships = UserRoleRelationship.objects.filter(user=create_test_user.id)
assert relationships.count() == 2
assert {rel.role.id for rel in relationships} == {
roles_fixture[1].id,
roles_fixture[2].id,
}
def test_destroy_relationship(
self, authenticated_client, roles_fixture, create_test_user
):
response = authenticated_client.delete(
reverse("user-roles-relationship", kwargs={"pk": create_test_user.id}),
)
assert response.status_code == status.HTTP_204_NO_CONTENT
relationships = UserRoleRelationship.objects.filter(role=roles_fixture[0].id)
assert relationships.count() == 0
def test_invalid_provider_group_id(self, authenticated_client, create_test_user):
invalid_id = "non-existent-id"
data = {"data": [{"type": "provider-group", "id": invalid_id}]}
response = authenticated_client.post(
reverse("user-roles-relationship", kwargs={"pk": create_test_user.id}),
data=data,
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
errors = response.json()["errors"][0]["detail"]
assert "valid UUID" in errors
@pytest.mark.django_db
class TestRoleProviderGroupRelationshipViewSet:
def test_create_relationship(
self, authenticated_client, roles_fixture, provider_groups_fixture
):
data = {
"data": [
{"type": "provider-group", "id": str(provider_group.id)}
for provider_group in provider_groups_fixture[:2]
]
}
response = authenticated_client.post(
reverse(
"role-provider-groups-relationship", kwargs={"pk": roles_fixture[0].id}
),
data=data,
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_204_NO_CONTENT
relationships = RoleProviderGroupRelationship.objects.filter(
role=roles_fixture[0].id
)
assert relationships.count() == 2
for relationship in relationships:
assert relationship.provider_group.id in [
pg.id for pg in provider_groups_fixture[:2]
]
def test_create_relationship_already_exists(
self, authenticated_client, roles_fixture, provider_groups_fixture
):
data = {
"data": [
{"type": "provider-group", "id": str(provider_group.id)}
for provider_group in provider_groups_fixture[:2]
]
}
authenticated_client.post(
reverse(
"role-provider-groups-relationship", kwargs={"pk": roles_fixture[0].id}
),
data=data,
content_type="application/vnd.api+json",
)
data = {
"data": [
{"type": "provider-group", "id": str(provider_groups_fixture[0].id)},
]
}
response = authenticated_client.post(
reverse(
"role-provider-groups-relationship", kwargs={"pk": roles_fixture[0].id}
),
data=data,
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
errors = response.json()["errors"]["detail"]
assert "already associated" in errors
def test_partial_update_relationship(
self, authenticated_client, roles_fixture, provider_groups_fixture
):
data = {
"data": [
{"type": "provider-group", "id": str(provider_groups_fixture[1].id)},
]
}
response = authenticated_client.patch(
reverse(
"role-provider-groups-relationship", kwargs={"pk": roles_fixture[2].id}
),
data=data,
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_204_NO_CONTENT
relationships = RoleProviderGroupRelationship.objects.filter(
role=roles_fixture[2].id
)
assert relationships.count() == 1
assert {rel.provider_group.id for rel in relationships} == {
provider_groups_fixture[1].id
}
data = {
"data": [
{"type": "provider-group", "id": str(provider_groups_fixture[1].id)},
{"type": "provider-group", "id": str(provider_groups_fixture[2].id)},
]
}
response = authenticated_client.patch(
reverse(
"role-provider-groups-relationship", kwargs={"pk": roles_fixture[2].id}
),
data=data,
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_204_NO_CONTENT
relationships = RoleProviderGroupRelationship.objects.filter(
role=roles_fixture[2].id
)
assert relationships.count() == 2
assert {rel.provider_group.id for rel in relationships} == {
provider_groups_fixture[1].id,
provider_groups_fixture[2].id,
}
def test_destroy_relationship(
self, authenticated_client, roles_fixture, provider_groups_fixture
):
response = authenticated_client.delete(
reverse(
"role-provider-groups-relationship", kwargs={"pk": roles_fixture[0].id}
),
)
assert response.status_code == status.HTTP_204_NO_CONTENT
relationships = RoleProviderGroupRelationship.objects.filter(
role=roles_fixture[0].id
)
assert relationships.count() == 0
def test_invalid_provider_group_id(self, authenticated_client, roles_fixture):
invalid_id = "non-existent-id"
data = {"data": [{"type": "provider-group", "id": invalid_id}]}
response = authenticated_client.post(
reverse(
"role-provider-groups-relationship", kwargs={"pk": roles_fixture[1].id}
),
data=data,
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
errors = response.json()["errors"][0]["detail"]
assert "valid UUID" in errors
@pytest.mark.django_db
class TestComplianceOverviewViewSet:
def test_compliance_overview_list_none(self, authenticated_client):
api/src/backend/api/v1/serializers.py
+301 -18
View File
@@ -27,6 +27,10 @@ from api.models import (
Finding,
ProviderSecret,
Invitation,
InvitationRoleRelationship,
Role,
RoleProviderGroupRelationship,
UserRoleRelationship,
ComplianceOverview,
)
from api.rls import Tenant
@@ -177,10 +181,26 @@ class UserSerializer(BaseSerializerV1):
"""
memberships = serializers.ResourceRelatedField(many=True, read_only=True)
roles = serializers.ResourceRelatedField(many=True, read_only=True)
class Meta:
model = User
fields = ["id", "name", "email", "company_name", "date_joined", "memberships"]
fields = [
"id",
"name",
"email",
"company_name",
"date_joined",
"memberships",
"roles",
]
extra_kwargs = {
"roles": {"read_only": True},
}
included_serializers = {
"roles": "api.v1.serializers.RoleSerializer",
}
class UserCreateSerializer(BaseWriteSerializer):
@@ -236,6 +256,73 @@ class UserUpdateSerializer(BaseWriteSerializer):
return super().update(instance, validated_data)
class RoleResourceIdentifierSerializer(serializers.Serializer):
resource_type = serializers.CharField(source="type")
id = serializers.UUIDField()
class JSONAPIMeta:
resource_name = "role-identifier"
def to_representation(self, instance):
"""
Ensure 'type' is used in the output instead of 'resource_type'.
"""
representation = super().to_representation(instance)
representation["type"] = representation.pop("resource_type", None)
return representation
def to_internal_value(self, data):
"""
Map 'type' back to 'resource_type' during input.
"""
data["resource_type"] = data.pop("type", None)
return super().to_internal_value(data)
class UserRoleRelationshipSerializer(RLSSerializer, BaseWriteSerializer):
"""
Serializer for modifying user memberships
"""
roles = serializers.ListField(
child=RoleResourceIdentifierSerializer(),
help_text="List of resource identifier objects representing roles.",
)
def create(self, validated_data):
role_ids = [item["id"] for item in validated_data["roles"]]
roles = Role.objects.filter(id__in=role_ids)
tenant_id = self.context.get("tenant_id")
new_relationships = [
UserRoleRelationship(
user=self.context.get("user"), role=r, tenant_id=tenant_id
)
for r in roles
]
UserRoleRelationship.objects.bulk_create(new_relationships)
return self.context.get("user")
def update(self, instance, validated_data):
role_ids = [item["id"] for item in validated_data["roles"]]
roles = Role.objects.filter(id__in=role_ids)
tenant_id = self.context.get("tenant_id")
instance.roles.clear()
new_relationships = [
UserRoleRelationship(user=instance, role=r, tenant_id=tenant_id)
for r in roles
]
UserRoleRelationship.objects.bulk_create(new_relationships)
return instance
class Meta:
model = UserRoleRelationship
fields = ["id", "roles"]
# Tasks
class TaskBase(serializers.ModelSerializer):
state_mapping = {
@@ -362,31 +449,30 @@ class ProviderGroupSerializer(RLSSerializer, BaseWriteSerializer):
providers = serializers.ResourceRelatedField(many=True, read_only=True)
def validate(self, attrs):
tenant = self.context["tenant_id"]
name = attrs.get("name", self.instance.name if self.instance else None)
# Exclude the current instance when checking for uniqueness during updates
queryset = ProviderGroup.objects.filter(tenant=tenant, name=name)
if self.instance:
queryset = queryset.exclude(pk=self.instance.pk)
if queryset.exists():
if ProviderGroup.objects.filter(name=attrs.get("name")).exists():
raise serializers.ValidationError(
{
"name": "A provider group with this name already exists for this tenant."
}
{"name": "A provider group with this name already exists."}
)
return super().validate(attrs)
class Meta:
model = ProviderGroup
fields = ["id", "name", "inserted_at", "updated_at", "providers", "url"]
read_only_fields = ["id", "inserted_at", "updated_at"]
fields = [
"id",
"name",
"inserted_at",
"updated_at",
"providers",
"roles",
"url",
]
extra_kwargs = {
"id": {"read_only": True},
"inserted_at": {"read_only": True},
"updated_at": {"read_only": True},
"roles": {"read_only": True},
"url": {"read_only": True},
}
@@ -465,6 +551,7 @@ class ProviderSerializer(RLSSerializer):
"connection",
# "scanner_args",
"secret",
"provider_groups",
"url",
]
@@ -1014,6 +1101,8 @@ class InvitationSerializer(RLSSerializer):
Serializer for the Invitation model.
"""
roles = serializers.ResourceRelatedField(many=True, queryset=Role.objects.all())
class Meta:
model = Invitation
fields = [
@@ -1023,6 +1112,7 @@ class InvitationSerializer(RLSSerializer):
"email",
"state",
"token",
"roles",
"expires_at",
"inviter",
"url",
@@ -1030,6 +1120,8 @@ class InvitationSerializer(RLSSerializer):
class InvitationBaseWriteSerializer(BaseWriteSerializer):
roles = serializers.ResourceRelatedField(many=True, queryset=Role.objects.all())
def validate_email(self, value):
user = User.objects.filter(email=value).first()
tenant_id = self.context["tenant_id"]
@@ -1066,31 +1158,54 @@ class InvitationCreateSerializer(InvitationBaseWriteSerializer, RLSSerializer):
class Meta:
model = Invitation
fields = ["email", "expires_at", "state", "token", "inviter"]
fields = ["email", "expires_at", "state", "token", "inviter", "roles"]
extra_kwargs = {
"token": {"read_only": True},
"state": {"read_only": True},
"inviter": {"read_only": True},
"expires_at": {"required": False},
"roles": {"required": False},
}
def create(self, validated_data):
inviter = self.context.get("request").user
tenant_id = self.context.get("tenant_id")
validated_data["inviter"] = inviter
return super().create(validated_data)
roles = validated_data.pop("roles", [])
invitation = super().create(validated_data)
for role in roles:
InvitationRoleRelationship.objects.create(
role=role, invitation=invitation, tenant_id=tenant_id
)
return invitation
class InvitationUpdateSerializer(InvitationBaseWriteSerializer):
class Meta:
model = Invitation
fields = ["id", "email", "expires_at", "state", "token"]
fields = ["id", "email", "expires_at", "state", "token", "roles"]
extra_kwargs = {
"token": {"read_only": True},
"state": {"read_only": True},
"expires_at": {"required": False},
"email": {"required": False},
"roles": {"required": False},
}
def update(self, instance, validated_data):
roles = validated_data.pop("roles", [])
tenant_id = self.context.get("tenant_id")
invitation = super().update(instance, validated_data)
if roles:
instance.roles.clear()
for role in roles:
InvitationRoleRelationship.objects.create(
role=role, invitation=invitation, tenant_id=tenant_id
)
return invitation
class InvitationAcceptSerializer(RLSSerializer):
"""Serializer for accepting an invitation."""
@@ -1102,6 +1217,174 @@ class InvitationAcceptSerializer(RLSSerializer):
fields = ["invitation_token"]
# Roles
class RoleSerializer(RLSSerializer, BaseWriteSerializer):
provider_groups = serializers.ResourceRelatedField(
many=True, queryset=ProviderGroup.objects.all()
)
def validate(self, attrs):
if Role.objects.filter(name=attrs.get("name")).exists():
raise serializers.ValidationError(
{"name": "A role with this name already exists."}
)
if attrs.get("manage_providers"):
attrs["unlimited_visibility"] = True
# Prevent updates to the admin role
if getattr(self.instance, "name", None) == "admin":
raise serializers.ValidationError(
{"name": "The admin role cannot be updated."}
)
return super().validate(attrs)
class Meta:
model = Role
fields = [
"id",
"name",
"manage_users",
"manage_account",
"manage_billing",
"manage_providers",
"manage_integrations",
"manage_scans",
"unlimited_visibility",
"inserted_at",
"updated_at",
"provider_groups",
"users",
"invitations",
"url",
]
extra_kwargs = {
"id": {"read_only": True},
"inserted_at": {"read_only": True},
"updated_at": {"read_only": True},
"users": {"read_only": True},
"url": {"read_only": True},
}
class RoleCreateSerializer(RoleSerializer):
def create(self, validated_data):
provider_groups = validated_data.pop("provider_groups", [])
users = validated_data.pop("users", [])
tenant_id = self.context.get("tenant_id")
role = Role.objects.create(tenant_id=tenant_id, **validated_data)
through_model_instances = [
RoleProviderGroupRelationship(
role=role,
provider_group=provider_group,
tenant_id=tenant_id,
)
for provider_group in provider_groups
]
RoleProviderGroupRelationship.objects.bulk_create(through_model_instances)
through_model_instances = [
UserRoleRelationship(
role=user,
user=user,
tenant_id=tenant_id,
)
for user in users
]
UserRoleRelationship.objects.bulk_create(through_model_instances)
return role
class RoleUpdateSerializer(RoleSerializer):
class Meta:
model = Role
fields = [
"id",
"name",
"manage_users",
"manage_account",
"manage_billing",
"manage_providers",
"manage_integrations",
"manage_scans",
"unlimited_visibility",
]
class ProviderGroupResourceIdentifierSerializer(serializers.Serializer):
resource_type = serializers.CharField(source="type")
id = serializers.UUIDField()
class JSONAPIMeta:
resource_name = "provider-group-identifier"
def to_representation(self, instance):
"""
Ensure 'type' is used in the output instead of 'resource_type'.
"""
representation = super().to_representation(instance)
representation["type"] = representation.pop("resource_type", None)
return representation
def to_internal_value(self, data):
"""
Map 'type' back to 'resource_type' during input.
"""
data["resource_type"] = data.pop("type", None)
return super().to_internal_value(data)
class RoleProviderGroupRelationshipSerializer(RLSSerializer, BaseWriteSerializer):
"""
Serializer for modifying role memberships
"""
provider_groups = serializers.ListField(
child=ProviderGroupResourceIdentifierSerializer(),
help_text="List of resource identifier objects representing provider groups.",
)
def create(self, validated_data):
provider_group_ids = [item["id"] for item in validated_data["provider_groups"]]
provider_groups = ProviderGroup.objects.filter(id__in=provider_group_ids)
tenant_id = self.context.get("tenant_id")
new_relationships = [
RoleProviderGroupRelationship(
role=self.context.get("role"), provider_group=pg, tenant_id=tenant_id
)
for pg in provider_groups
]
RoleProviderGroupRelationship.objects.bulk_create(new_relationships)
return self.context.get("role")
def update(self, instance, validated_data):
provider_group_ids = [item["id"] for item in validated_data["provider_groups"]]
provider_groups = ProviderGroup.objects.filter(id__in=provider_group_ids)
tenant_id = self.context.get("tenant_id")
instance.provider_groups.clear()
new_relationships = [
RoleProviderGroupRelationship(
role=instance, provider_group=pg, tenant_id=tenant_id
)
for pg in provider_groups
]
RoleProviderGroupRelationship.objects.bulk_create(new_relationships)
return instance
class Meta:
model = RoleProviderGroupRelationship
fields = ["id", "provider_groups"]
# Compliance overview
api/src/backend/api/v1/urls.py
+18
View File
@@ -19,6 +19,9 @@ from api.v1.views import (
ProviderSecretViewSet,
InvitationViewSet,
InvitationAcceptViewSet,
RoleViewSet,
RoleProviderGroupRelationshipView,
UserRoleRelationshipView,
OverviewViewSet,
ComplianceOverviewViewSet,
)
@@ -33,6 +36,7 @@ router.register(r"scans", ScanViewSet, basename="scan")
router.register(r"tasks", TaskViewSet, basename="task")
router.register(r"resources", ResourceViewSet, basename="resource")
router.register(r"findings", FindingViewSet, basename="finding")
router.register(r"roles", RoleViewSet, basename="role")
router.register(
r"compliance-overviews", ComplianceOverviewViewSet, basename="complianceoverview"
)
@@ -78,6 +82,20 @@ urlpatterns = [
InvitationAcceptViewSet.as_view({"post": "accept"}),
name="invitation-accept",
),
path(
"roles/<uuid:pk>/relationships/provider_groups",
RoleProviderGroupRelationshipView.as_view(
{"post": "create", "patch": "partial_update", "delete": "destroy"}
),
name="role-provider-groups-relationship",
),
path(
"users/<uuid:pk>/relationships/roles",
UserRoleRelationshipView.as_view(
{"post": "create", "patch": "partial_update", "delete": "destroy"}
),
name="user-roles-relationship",
),
path("", include(router.urls)),
path("", include(tenants_router.urls)),
path("", include(users_router.urls)),
api/src/backend/api/v1/views.py
+490 -15
View File
@@ -7,6 +7,7 @@ from django.urls import reverse
from django.utils.decorators import method_decorator
from django.views.decorators.cache import cache_control
from drf_spectacular.settings import spectacular_settings
from drf_spectacular_jsonapi.schemas.openapi import JsonApiAutoSchema
from drf_spectacular.utils import (
extend_schema,
extend_schema_view,
@@ -24,9 +25,10 @@ from rest_framework.exceptions import (
ValidationError,
)
from rest_framework.generics import get_object_or_404, GenericAPIView
from rest_framework_json_api.views import Response
from rest_framework_json_api.views import RelationshipView, Response
from rest_framework_simplejwt.exceptions import InvalidToken
from rest_framework_simplejwt.exceptions import TokenError
from rest_framework.permissions import SAFE_METHODS
from api.base_views import BaseTenantViewset, BaseRLSViewSet, BaseUserViewset
from api.db_router import MainRouter
@@ -42,11 +44,13 @@ from api.filters import (
ProviderSecretFilter,
InvitationFilter,
UserFilter,
RoleFilter,
ComplianceOverviewFilter,
)
from api.models import (
StatusChoices,
User,
UserRoleRelationship,
Membership,
Provider,
ProviderGroup,
@@ -57,9 +61,12 @@ from api.models import (
Finding,
ProviderSecret,
Invitation,
Role,
RoleProviderGroupRelationship,
ComplianceOverview,
)
from api.pagination import ComplianceOverviewPagination
from api.rbac.permissions import DISABLE_RBAC, HasPermissions, Permissions
from api.rls import Tenant
from api.utils import validate_invitation
from api.uuid_utils import datetime_to_uuid7
@@ -69,10 +76,12 @@ from api.v1.serializers import (
UserSerializer,
UserCreateSerializer,
UserUpdateSerializer,
UserRoleRelationshipSerializer,
MembershipSerializer,
ProviderGroupSerializer,
ProviderGroupUpdateSerializer,
ProviderGroupMembershipUpdateSerializer,
RoleProviderGroupRelationshipSerializer,
ProviderSerializer,
ProviderCreateSerializer,
ProviderUpdateSerializer,
@@ -90,6 +99,9 @@ from api.v1.serializers import (
InvitationCreateSerializer,
InvitationUpdateSerializer,
InvitationAcceptSerializer,
RoleSerializer,
RoleCreateSerializer,
RoleUpdateSerializer,
ComplianceOverviewSerializer,
ComplianceOverviewFullSerializer,
OverviewProviderSerializer,
@@ -107,6 +119,11 @@ CACHE_DECORATOR = cache_control(
)
class RelationshipViewSchema(JsonApiAutoSchema):
def _resolve_path_parameters(self, _path_variables):
return []
@extend_schema(
tags=["Token"],
summary="Obtain a token",
@@ -261,6 +278,26 @@ class UserViewSet(BaseUserViewset):
filterset_class = UserFilter
ordering = ["-date_joined"]
ordering_fields = ["name", "email", "company_name", "date_joined", "is_active"]
required_permissions = [Permissions.MANAGE_USERS]
permission_classes = BaseRLSViewSet.permission_classes + [HasPermissions]
def initial(self, request, *args, **kwargs):
"""
Sets required_permissions before permissions are checked.
"""
self.required_permissions = self.get_required_permissions()
super().initial(request, *args, **kwargs)
def get_required_permissions(self):
"""
Returns the required permissions based on the request method.
"""
if self.action == "me":
# No permissions required for me request
return []
else:
# Require permission for the rest of the requests
return [Permissions.MANAGE_USERS]
def get_queryset(self):
# If called during schema generation, return an empty queryset
@@ -337,11 +374,124 @@ class UserViewSet(BaseUserViewset):
user=user, tenant=tenant, role=role
)
if invitation:
# TODO: Add roles to output relationships
user_role = []
for role in invitation.roles.all():
user_role.append(
UserRoleRelationship.objects.using(MainRouter.admin_db).create(
user=user, role=role, tenant=invitation.tenant
)
)
invitation.state = Invitation.State.ACCEPTED
invitation.save(using=MainRouter.admin_db)
else:
role = Role.objects.using(MainRouter.admin_db).create(
name="admin",
tenant_id=tenant.id,
manage_users=True,
manage_account=True,
manage_billing=True,
manage_providers=True,
manage_integrations=True,
manage_scans=True,
unlimited_visibility=True,
)
UserRoleRelationship.objects.using(MainRouter.admin_db).create(
user=user,
role=role,
tenant_id=tenant.id,
)
return Response(data=UserSerializer(user).data, status=status.HTTP_201_CREATED)
@extend_schema_view(
create=extend_schema(
tags=["User"],
summary="Create a new user-roles relationship",
description="Add a new user-roles relationship to the system by providing the required user-roles details.",
responses={
204: OpenApiResponse(description="Relationship created successfully"),
400: OpenApiResponse(
description="Bad request (e.g., relationship already exists)"
),
},
),
partial_update=extend_schema(
tags=["User"],
summary="Partially update a user-roles relationship",
description="Update the user-roles relationship information without affecting other fields.",
responses={
204: OpenApiResponse(
response=None, description="Relationship updated successfully"
)
},
),
destroy=extend_schema(
tags=["User"],
summary="Delete a user-roles relationship",
description="Remove the user-roles relationship from the system by their ID.",
responses={
204: OpenApiResponse(
response=None, description="Relationship deleted successfully"
)
},
),
)
class UserRoleRelationshipView(RelationshipView, BaseRLSViewSet):
queryset = User.objects.all()
serializer_class = UserRoleRelationshipSerializer
resource_name = "roles"
http_method_names = ["post", "patch", "delete"]
schema = RelationshipViewSchema()
def get_queryset(self):
return User.objects.all()
def create(self, request, *args, **kwargs):
user = self.get_object()
role_ids = [item["id"] for item in request.data]
existing_relationships = UserRoleRelationship.objects.filter(
user=user, role_id__in=role_ids
)
if existing_relationships.exists():
return Response(
{"detail": "One or more roles are already associated with the user."},
status=status.HTTP_400_BAD_REQUEST,
)
serializer = self.get_serializer(
data={"roles": request.data},
context={
"user": user,
"tenant_id": self.request.tenant_id,
"request": request,
},
)
serializer.is_valid(raise_exception=True)
serializer.save()
return Response(status=status.HTTP_204_NO_CONTENT)
def partial_update(self, request, *args, **kwargs):
user = self.get_object()
serializer = self.get_serializer(
instance=user,
data={"roles": request.data},
context={"tenant_id": self.request.tenant_id, "request": request},
)
serializer.is_valid(raise_exception=True)
serializer.save()
return Response(status=status.HTTP_204_NO_CONTENT)
def destroy(self, request, *args, **kwargs):
user = self.get_object()
user.roles.clear()
return Response(status=status.HTTP_204_NO_CONTENT)
@extend_schema_view(
list=extend_schema(
tags=["Tenant"],
@@ -379,6 +529,8 @@ class TenantViewSet(BaseTenantViewset):
search_fields = ["name"]
ordering = ["-inserted_at"]
ordering_fields = ["name", "inserted_at", "updated_at"]
required_permissions = [Permissions.MANAGE_ACCOUNT]
permission_classes = BaseRLSViewSet.permission_classes + [HasPermissions]
def get_queryset(self):
return Tenant.objects.all()
@@ -535,9 +687,46 @@ class ProviderGroupViewSet(BaseRLSViewSet):
filterset_class = ProviderGroupFilter
http_method_names = ["get", "post", "patch", "put", "delete"]
ordering = ["inserted_at"]
required_permissions = []
permission_classes = BaseRLSViewSet.permission_classes + [HasPermissions]
def initial(self, request, *args, **kwargs):
"""
Sets required_permissions before permissions are checked.
"""
self.required_permissions = self.get_required_permissions()
super().initial(request, *args, **kwargs)
def get_required_permissions(self):
"""
Returns the required permissions based on the request method.
"""
if DISABLE_RBAC or self.request.method in SAFE_METHODS:
# No permissions required for GET requests
return []
else:
# Require permission for non-GET requests
return [Permissions.MANAGE_PROVIDERS]
def get_queryset(self):
return ProviderGroup.objects.prefetch_related("providers")
user = self.request.user
user_roles = user.roles.all()
# Check if any of the user's roles have UNLIMITED_VISIBILITY
if DISABLE_RBAC or getattr(
user_roles[0], Permissions.UNLIMITED_VISIBILITY.value, False
):
# User has unlimited visibility, return all provider groups
return ProviderGroup.objects.prefetch_related("providers")
# Collect provider groups associated with the user's roles
provider_groups = (
ProviderGroup.objects.filter(roles__in=user_roles)
.distinct()
.prefetch_related("providers")
)
return provider_groups
def get_serializer_class(self):
if self.action == "partial_update":
@@ -642,9 +831,43 @@ class ProviderViewSet(BaseRLSViewSet):
"inserted_at",
"updated_at",
]
required_permissions = []
permission_classes = BaseRLSViewSet.permission_classes + [HasPermissions]
def initial(self, request, *args, **kwargs):
"""
Sets required_permissions before permissions are checked.
"""
self.required_permissions = self.get_required_permissions()
super().initial(request, *args, **kwargs)
def get_required_permissions(self):
"""
Returns the required permissions based on the request method.
"""
if DISABLE_RBAC or self.request.method in SAFE_METHODS:
# No permissions required for GET requests
return []
else:
# Require permission for non-GET requests
return [Permissions.MANAGE_PROVIDERS]
def get_queryset(self):
return Provider.objects.all()
user = self.request.user
user_roles = user.roles.all()
if DISABLE_RBAC or getattr(
user_roles[0], Permissions.UNLIMITED_VISIBILITY.value, False
):
# User has unlimited visibility, return all providers
return Provider.objects.all()
# User lacks permission, filter providers based on provider groups associated with the role
provider_groups = user_roles[0].provider_groups.all()
providers = Provider.objects.filter(
provider_groups__in=provider_groups
).distinct()
return providers
def get_serializer_class(self):
if self.action == "create":
@@ -769,9 +992,42 @@ class ScanViewSet(BaseRLSViewSet):
"inserted_at",
"updated_at",
]
required_permissions = [Permissions.MANAGE_SCANS]
permission_classes = BaseRLSViewSet.permission_classes + [HasPermissions]
def initial(self, request, *args, **kwargs):
"""
Sets required_permissions before permissions are checked.
"""
self.required_permissions = self.get_required_permissions()
super().initial(request, *args, **kwargs)
def get_required_permissions(self):
"""
Returns the required permissions based on the request method.
"""
if DISABLE_RBAC or self.request.method in SAFE_METHODS:
# No permissions required for GET requests
return []
else:
# Require permission for non-GET requests
return [Permissions.MANAGE_SCANS]
def get_queryset(self):
return Scan.objects.all()
user = self.request.user
user_roles = user.roles.all()
if DISABLE_RBAC or getattr(
user_roles[0], Permissions.UNLIMITED_VISIBILITY.value, False
):
# User has unlimited visibility, return all scans
return Scan.objects.all()
# User lacks permission, filter providers based on provider groups associated with the role
provider_groups = user_roles[0].provider_groups.all()
providers = Provider.objects.filter(
provider_groups__in=provider_groups
).distinct()
return Scan.objects.filter(provider__in=providers).distinct()
def get_serializer_class(self):
if self.action == "create":
@@ -855,11 +1111,28 @@ class TaskViewSet(BaseRLSViewSet):
search_fields = ["name"]
ordering = ["-inserted_at"]
ordering_fields = ["inserted_at", "completed_at", "name", "state"]
required_permissions = []
permission_classes = BaseRLSViewSet.permission_classes + [HasPermissions]
def get_queryset(self):
return Task.objects.annotate(
name=F("task_runner_task__task_name"), state=F("task_runner_task__status")
)
user = self.request.user
user_roles = user.roles.all()
if DISABLE_RBAC or getattr(
user_roles[0], Permissions.UNLIMITED_VISIBILITY.value, False
):
# User has unlimited visibility, return all tasks
return Task.objects.annotate(
name=F("task_runner_task__task_name"),
state=F("task_runner_task__status"),
)
# User lacks permission, filter tasks based on provider groups associated with the role
provider_groups = user_roles[0].provider_groups.all()
providers = Provider.objects.filter(
provider_groups__in=provider_groups
).distinct()
scans = Scan.objects.filter(provider__in=providers).distinct()
return Task.objects.filter(scan__in=scans).distinct()
def destroy(self, request, *args, pk=None, **kwargs):
task = get_object_or_404(Task, pk=pk)
@@ -920,11 +1193,33 @@ class ResourceViewSet(BaseRLSViewSet):
"inserted_at",
"updated_at",
]
required_permissions = []
permission_classes = BaseRLSViewSet.permission_classes + [HasPermissions]
def initial(self, request, *args, **kwargs):
"""
Sets required_permissions before permissions are checked.
"""
self.required_permissions = ResourceViewSet.required_permissions
super().initial(request, *args, **kwargs)
def get_queryset(self):
queryset = Resource.objects.all()
search_value = self.request.query_params.get("filter[search]", None)
user = self.request.user
user_roles = user.roles.all()
if DISABLE_RBAC or getattr(
user_roles[0], Permissions.UNLIMITED_VISIBILITY.value, False
):
# User has unlimited visibility, return all scans
queryset = Resource.objects.all()
else:
# User lacks permission, filter providers based on provider groups associated with the role
provider_groups = user_roles[0].provider_groups.all()
providers = Provider.objects.filter(
provider_groups__in=provider_groups
).distinct()
queryset = Resource.objects.filter(provider__in=providers).distinct()
search_value = self.request.query_params.get("filter[search]", None)
if search_value:
# Django's ORM will build a LEFT JOIN and OUTER JOIN on the "through" table, resulting in duplicates
# The duplicates then require a `distinct` query
@@ -988,16 +1283,34 @@ class FindingViewSet(BaseRLSViewSet):
"inserted_at",
"updated_at",
]
required_permissions = []
permission_classes = BaseRLSViewSet.permission_classes + [HasPermissions]
def inserted_at_to_uuidv7(self, inserted_at):
if inserted_at is None:
return None
return datetime_to_uuid7(inserted_at)
def initial(self, request, *args, **kwargs):
"""
Sets required_permissions before permissions are checked.
"""
self.required_permissions = ResourceViewSet.required_permissions
super().initial(request, *args, **kwargs)
def get_queryset(self):
queryset = Finding.objects.all()
search_value = self.request.query_params.get("filter[search]", None)
user = self.request.user
user_roles = user.roles.all()
if DISABLE_RBAC or getattr(
user_roles[0], Permissions.UNLIMITED_VISIBILITY.value, False
):
# User has unlimited visibility, return all scans
queryset = Finding.objects.all()
else:
# User lacks permission, filter providers based on provider groups associated with the role
provider_groups = user_roles[0].provider_groups.all()
providers = Provider.objects.filter(
provider_groups__in=provider_groups
).distinct()
scans = Scan.objects.filter(provider__in=providers).distinct()
queryset = Finding.objects.filter(scan__in=scans).distinct()
search_value = self.request.query_params.get("filter[search]", None)
if search_value:
# Django's ORM will build a LEFT JOIN and OUTER JOIN on any "through" tables, resulting in duplicates
# The duplicates then require a `distinct` query
@@ -1025,6 +1338,11 @@ class FindingViewSet(BaseRLSViewSet):
return queryset
def inserted_at_to_uuidv7(self, inserted_at):
if inserted_at is None:
return None
return datetime_to_uuid7(inserted_at)
@extend_schema_view(
list=extend_schema(
@@ -1124,6 +1442,8 @@ class InvitationViewSet(BaseRLSViewSet):
"state",
"inviter",
]
required_permissions = [Permissions.MANAGE_ACCOUNT]
permission_classes = BaseRLSViewSet.permission_classes + [HasPermissions]
def get_queryset(self):
return Invitation.objects.all()
@@ -1211,6 +1531,14 @@ class InvitationAcceptViewSet(BaseRLSViewSet):
user=user,
tenant=invitation.tenant,
)
# TODO: Add roles to output relationships
user_role = []
for role in invitation.roles.all():
user_role.append(
UserRoleRelationship.objects.using(MainRouter.admin_db).create(
user=user, role=role, tenant=invitation.tenant
)
)
invitation.state = Invitation.State.ACCEPTED
invitation.save(using=MainRouter.admin_db)
@@ -1219,6 +1547,153 @@ class InvitationAcceptViewSet(BaseRLSViewSet):
return Response(data=membership_serializer.data, status=status.HTTP_201_CREATED)
@extend_schema(tags=["Role"])
@extend_schema_view(
list=extend_schema(
tags=["Role"],
summary="List all roles",
description="Retrieve a list of all roles with options for filtering by various criteria.",
),
retrieve=extend_schema(
tags=["Role"],
summary="Retrieve data from a role",
description="Fetch detailed information about a specific role by their ID.",
),
create=extend_schema(
tags=["Role"],
summary="Create a new role",
description="Add a new role to the system by providing the required role details.",
),
partial_update=extend_schema(
tags=["Role"],
summary="Partially update a role",
description="Update certain fields of an existing role's information without affecting other fields.",
responses={200: RoleSerializer},
),
destroy=extend_schema(
tags=["Role"],
summary="Delete a role",
description="Remove a role from the system by their ID.",
),
)
class RoleViewSet(BaseRLSViewSet):
queryset = Role.objects.all()
serializer_class = RoleSerializer
filterset_class = RoleFilter
http_method_names = ["get", "post", "patch", "delete"]
ordering = ["inserted_at"]
required_permissions = [Permissions.MANAGE_ACCOUNT]
permission_classes = BaseRLSViewSet.permission_classes + [HasPermissions]
def get_queryset(self):
return Role.objects.all()
def get_serializer_class(self):
if self.action == "create":
return RoleCreateSerializer
elif self.action == "partial_update":
return RoleUpdateSerializer
return super().get_serializer_class()
def partial_update(self, request, *args, **kwargs):
user = request.user
user_role = user.roles.all().first()
# If the user is the owner of the role, the manage_account field is not editable
if user_role and kwargs["pk"] == str(user_role.id):
request.data["manage_account"] = str(user_role.manage_account).lower()
return super().partial_update(request, *args, **kwargs)
@extend_schema_view(
create=extend_schema(
tags=["Role"],
summary="Create a new role-provider_groups relationship",
description="Add a new role-provider_groups relationship to the system by providing the required role-provider_groups details.",
responses={
204: OpenApiResponse(description="Relationship created successfully"),
400: OpenApiResponse(
description="Bad request (e.g., relationship already exists)"
),
},
),
partial_update=extend_schema(
tags=["Role"],
summary="Partially update a role-provider_groups relationship",
description="Update the role-provider_groups relationship information without affecting other fields.",
responses={
204: OpenApiResponse(
response=None, description="Relationship updated successfully"
)
},
),
destroy=extend_schema(
tags=["Role"],
summary="Delete a role-provider_groups relationship",
description="Remove the role-provider_groups relationship from the system by their ID.",
responses={
204: OpenApiResponse(
response=None, description="Relationship deleted successfully"
)
},
),
)
class RoleProviderGroupRelationshipView(RelationshipView, BaseRLSViewSet):
queryset = Role.objects.all()
serializer_class = RoleProviderGroupRelationshipSerializer
resource_name = "provider_groups"
http_method_names = ["post", "patch", "delete"]
schema = RelationshipViewSchema()
def get_queryset(self):
return Role.objects.all()
def create(self, request, *args, **kwargs):
role = self.get_object()
provider_group_ids = [item["id"] for item in request.data]
existing_relationships = RoleProviderGroupRelationship.objects.filter(
role=role, provider_group_id__in=provider_group_ids
)
if existing_relationships.exists():
return Response(
{
"detail": "One or more provider groups are already associated with the role."
},
status=status.HTTP_400_BAD_REQUEST,
)
serializer = self.get_serializer(
data={"provider_groups": request.data},
context={
"role": role,
"tenant_id": self.request.tenant_id,
"request": request,
},
)
serializer.is_valid(raise_exception=True)
serializer.save()
return Response(status=status.HTTP_204_NO_CONTENT)
def partial_update(self, request, *args, **kwargs):
role = self.get_object()
serializer = self.get_serializer(
instance=role,
data={"provider_groups": request.data},
context={"tenant_id": self.request.tenant_id, "request": request},
)
serializer.is_valid(raise_exception=True)
serializer.save()
return Response(status=status.HTTP_204_NO_CONTENT)
def destroy(self, request, *args, **kwargs):
role = self.get_object()
role.provider_groups.clear()
return Response(status=status.HTTP_204_NO_CONTENT)
@extend_schema_view(
list=extend_schema(
tags=["Compliance Overview"],
api/src/backend/config/django/base.py
+3
View File
@@ -207,3 +207,6 @@ CACHE_STALE_WHILE_REVALIDATE = env.int("DJANGO_STALE_WHILE_REVALIDATE", 60)
TESTING = False
# Disable RBAC during tests/demos
DISABLE_RBAC = False
api/src/backend/conftest.py
+155
View File
@@ -10,6 +10,7 @@ from prowler.lib.check.models import Severity
from prowler.lib.outputs.finding import Status
from rest_framework import status
from rest_framework.test import APIClient
from unittest.mock import patch
from api.models import (
Finding,
@@ -20,6 +21,7 @@ from api.models import (
ProviderGroup,
Resource,
ResourceTag,
Role,
Scan,
StateChoices,
Task,
@@ -27,6 +29,7 @@ from api.models import (
ProviderSecret,
Invitation,
ComplianceOverview,
UserRoleRelationship,
)
from api.rls import Tenant
from api.v1.serializers import TokenSerializer
@@ -72,6 +75,16 @@ def disable_logging():
logging.disable(logging.CRITICAL)
@pytest.fixture(scope="function")
def patch_testing_flag():
"""
Fixture to patch the TESTING flag to True during tests.
"""
with patch("api.rbac.permissions.DISABLE_RBAC", True):
with patch("api.v1.views.DISABLE_RBAC", True):
yield
@pytest.fixture(scope="session", autouse=True)
def create_test_user(django_db_setup, django_db_blocker):
with django_db_blocker.unblock():
@@ -83,6 +96,106 @@ def create_test_user(django_db_setup, django_db_blocker):
return user
@pytest.fixture(scope="function")
def create_test_user_rbac(django_db_setup, django_db_blocker):
with django_db_blocker.unblock():
user = User.objects.create_user(
name="testing",
email="rbac@rbac.com",
password=TEST_PASSWORD,
)
tenant = Tenant.objects.create(
name="Tenant Test",
)
Membership.objects.create(
user=user,
tenant=tenant,
role=Membership.RoleChoices.OWNER,
)
Role.objects.create(
name="admin",
tenant_id=tenant.id,
manage_users=True,
manage_account=True,
manage_billing=True,
manage_providers=True,
manage_integrations=True,
manage_scans=True,
unlimited_visibility=True,
)
UserRoleRelationship.objects.create(
user=user,
role=Role.objects.get(name="admin"),
tenant_id=tenant.id,
)
return user
@pytest.fixture(scope="function")
def create_test_user_rbac_limited(django_db_setup, django_db_blocker):
with django_db_blocker.unblock():
user = User.objects.create_user(
name="testing_limited",
email="rbac_limited@rbac.com",
password=TEST_PASSWORD,
)
tenant = Tenant.objects.create(
name="Tenant Test",
)
Membership.objects.create(
user=user,
tenant=tenant,
role=Membership.RoleChoices.OWNER,
)
Role.objects.create(
name="limited",
tenant_id=tenant.id,
manage_users=False,
manage_account=False,
manage_billing=False,
manage_providers=False,
manage_integrations=False,
manage_scans=False,
unlimited_visibility=False,
)
UserRoleRelationship.objects.create(
user=user,
role=Role.objects.get(name="limited"),
tenant_id=tenant.id,
)
return user
@pytest.fixture
def authenticated_client_rbac(create_test_user_rbac, tenants_fixture, client):
client.user = create_test_user_rbac
serializer = TokenSerializer(
data={"type": "tokens", "email": "rbac@rbac.com", "password": TEST_PASSWORD}
)
serializer.is_valid()
access_token = serializer.validated_data["access"]
client.defaults["HTTP_AUTHORIZATION"] = f"Bearer {access_token}"
return client
@pytest.fixture
def authenticated_client_no_permissions_rbac(
create_test_user_rbac_limited, tenants_fixture, client
):
client.user = create_test_user_rbac_limited
serializer = TokenSerializer(
data={
"type": "tokens",
"email": "rbac_limited@rbac.com",
"password": TEST_PASSWORD,
}
)
serializer.is_valid()
access_token = serializer.validated_data["access"]
client.defaults["HTTP_AUTHORIZATION"] = f"Bearer {access_token}"
return client
@pytest.fixture
def authenticated_client(create_test_user, tenants_fixture, client):
client.user = create_test_user
@@ -104,6 +217,7 @@ def authenticated_api_client(create_test_user, tenants_fixture):
serializer.is_valid()
access_token = serializer.validated_data["access"]
client.defaults["HTTP_AUTHORIZATION"] = f"Bearer {access_token}"
return client
@@ -128,6 +242,7 @@ def tenants_fixture(create_test_user):
tenant3 = Tenant.objects.create(
name="Tenant Three",
)
return tenant1, tenant2, tenant3
@@ -210,6 +325,46 @@ def provider_groups_fixture(tenants_fixture):
return pgroup1, pgroup2, pgroup3
@pytest.fixture
def roles_fixture(tenants_fixture):
tenant, *_ = tenants_fixture
role1 = Role.objects.create(
name="Role One",
tenant_id=tenant.id,
manage_users=True,
manage_account=True,
manage_billing=True,
manage_providers=True,
manage_integrations=False,
manage_scans=True,
unlimited_visibility=False,
)
role2 = Role.objects.create(
name="Role Two",
tenant_id=tenant.id,
manage_users=False,
manage_account=False,
manage_billing=False,
manage_providers=True,
manage_integrations=True,
manage_scans=True,
unlimited_visibility=True,
)
role3 = Role.objects.create(
name="Role Three",
tenant_id=tenant.id,
manage_users=True,
manage_account=True,
manage_billing=True,
manage_providers=True,
manage_integrations=True,
manage_scans=True,
unlimited_visibility=True,
)
return role1, role2, role3
@pytest.fixture
def provider_secret_fixture(providers_fixture):
return tuple(