9dde1805aa
# What this PR does - Adds [`mypy` static type checking](https://mypy-lang.org/) to our CI pipeline. Currently there is still a **ton** of errors being returned by the tool, as we'll need to fix pre-existing errors. I think we can slowly chip away at these errors in small PRs, doing them all in one large PR is likely very risky. - Also, this PR starts chipping away at one of the main type errors that we have which is accessing the `datetime` class (from the `datetime` library) or `timedelta` function on the `django.utils.timezone` module. Basically we should be instead accessing these two objects from the native `datetime` module. This makes sense because the [`__all__` attribute](https://github.com/django/django/blob/main/django/utils/timezone.py#L14-L30) in `django.utils.timezone` does not re-export `datetime` or `timedelta`. - splits `engine` dependencies out into `requirements.txt` and `requirements-dev.txt` ## Checklist - [ ] Unit, integration, and e2e (if applicable) tests updated (N/A) - [ ] Documentation added (or `pr:no public docs` PR label added if not required) (N/A) - [ ] `CHANGELOG.md` updated (or `pr:no changelog` PR label added if not required) (N/A)
428 lines
16 KiB
Python
428 lines
16 KiB
Python
import typing
|
|
|
|
import pytest
|
|
from rest_framework.views import APIView
|
|
from rest_framework.viewsets import ViewSetMixin
|
|
|
|
from apps.api.permissions import (
|
|
RBAC_PERMISSIONS_ATTR,
|
|
GrafanaAPIPermission,
|
|
HasRBACPermissions,
|
|
IsOwner,
|
|
IsOwnerOrHasRBACPermissions,
|
|
LegacyAccessControlCompatiblePermission,
|
|
RBACObjectPermissionsAttribute,
|
|
RBACPermission,
|
|
RBACPermissionsAttribute,
|
|
get_most_authorized_role,
|
|
user_is_authorized,
|
|
)
|
|
|
|
|
|
class MockedOrg:
|
|
def __init__(self, org_has_rbac_enabled: bool) -> None:
|
|
self.is_rbac_permissions_enabled = org_has_rbac_enabled
|
|
|
|
|
|
class MockedUser:
|
|
def __init__(
|
|
self, permissions: typing.List[LegacyAccessControlCompatiblePermission], org_has_rbac_enabled=True
|
|
) -> None:
|
|
self.permissions = [GrafanaAPIPermission(action=perm.value) for perm in permissions]
|
|
self.role = get_most_authorized_role(permissions)
|
|
self.organization = MockedOrg(org_has_rbac_enabled)
|
|
|
|
|
|
class MockedSchedule:
|
|
def __init__(self, user: MockedUser) -> None:
|
|
self.user = user
|
|
|
|
|
|
class MockedRequest:
|
|
def __init__(self, user: typing.Optional[MockedUser] = None, method: typing.Optional[str] = None) -> None:
|
|
if user:
|
|
self.user = user
|
|
if method:
|
|
self.method = method
|
|
|
|
|
|
class MockedViewSet(ViewSetMixin):
|
|
def __init__(
|
|
self,
|
|
action: str,
|
|
rbac_permissions: typing.Optional[RBACPermissionsAttribute] = None,
|
|
rbac_object_permissions: typing.Optional[RBACObjectPermissionsAttribute] = None,
|
|
) -> None:
|
|
super().__init__()
|
|
self.action = action
|
|
|
|
if rbac_permissions:
|
|
self.rbac_permissions = rbac_permissions
|
|
if rbac_object_permissions:
|
|
self.rbac_object_permissions = rbac_object_permissions
|
|
|
|
|
|
class MockedAPIView(APIView):
|
|
def __init__(
|
|
self,
|
|
rbac_permissions: typing.Optional[RBACPermissionsAttribute] = None,
|
|
rbac_object_permissions: typing.Optional[RBACObjectPermissionsAttribute] = None,
|
|
) -> None:
|
|
super().__init__()
|
|
|
|
if rbac_permissions:
|
|
self.rbac_permissions = rbac_permissions
|
|
if rbac_object_permissions:
|
|
self.rbac_object_permissions = rbac_object_permissions
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"user_permissions,required_permissions,org_has_rbac_enabled,expected_result",
|
|
[
|
|
(
|
|
[RBACPermission.Permissions.ALERT_GROUPS_READ],
|
|
[RBACPermission.Permissions.ALERT_GROUPS_READ],
|
|
True,
|
|
True,
|
|
),
|
|
(
|
|
[RBACPermission.Permissions.ALERT_GROUPS_READ],
|
|
[RBACPermission.Permissions.ALERT_GROUPS_READ],
|
|
False,
|
|
True,
|
|
),
|
|
(
|
|
[RBACPermission.Permissions.ALERT_GROUPS_READ, RBACPermission.Permissions.ALERT_GROUPS_WRITE],
|
|
[RBACPermission.Permissions.ALERT_GROUPS_READ, RBACPermission.Permissions.ALERT_GROUPS_WRITE],
|
|
True,
|
|
True,
|
|
),
|
|
(
|
|
[RBACPermission.Permissions.ALERT_GROUPS_READ, RBACPermission.Permissions.ALERT_GROUPS_WRITE],
|
|
[RBACPermission.Permissions.ALERT_GROUPS_READ, RBACPermission.Permissions.ALERT_GROUPS_WRITE],
|
|
False,
|
|
True,
|
|
),
|
|
(
|
|
[RBACPermission.Permissions.ALERT_GROUPS_WRITE],
|
|
[RBACPermission.Permissions.ALERT_GROUPS_READ],
|
|
True,
|
|
False,
|
|
),
|
|
(
|
|
[RBACPermission.Permissions.ALERT_GROUPS_WRITE],
|
|
[RBACPermission.Permissions.ALERT_GROUPS_READ],
|
|
False,
|
|
True,
|
|
),
|
|
(
|
|
[RBACPermission.Permissions.ALERT_GROUPS_READ],
|
|
[RBACPermission.Permissions.ALERT_GROUPS_READ, RBACPermission.Permissions.ALERT_GROUPS_WRITE],
|
|
False,
|
|
False,
|
|
),
|
|
(
|
|
[RBACPermission.Permissions.ALERT_GROUPS_READ],
|
|
[RBACPermission.Permissions.ALERT_GROUPS_READ, RBACPermission.Permissions.ALERT_GROUPS_WRITE],
|
|
True,
|
|
False,
|
|
),
|
|
],
|
|
)
|
|
def test_user_is_authorized(user_permissions, required_permissions, org_has_rbac_enabled, expected_result) -> None:
|
|
user = MockedUser(user_permissions, org_has_rbac_enabled=org_has_rbac_enabled)
|
|
assert user_is_authorized(user, required_permissions) == expected_result
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"permissions,expected_role",
|
|
[
|
|
([RBACPermission.Permissions.ALERT_GROUPS_READ], RBACPermission.Permissions.ALERT_GROUPS_READ.fallback_role),
|
|
(
|
|
[RBACPermission.Permissions.ALERT_GROUPS_READ, RBACPermission.Permissions.ALERT_GROUPS_WRITE],
|
|
RBACPermission.Permissions.ALERT_GROUPS_WRITE.fallback_role,
|
|
),
|
|
(
|
|
[
|
|
RBACPermission.Permissions.USER_SETTINGS_READ,
|
|
RBACPermission.Permissions.USER_SETTINGS_WRITE,
|
|
RBACPermission.Permissions.USER_SETTINGS_ADMIN,
|
|
],
|
|
RBACPermission.Permissions.USER_SETTINGS_ADMIN.fallback_role,
|
|
),
|
|
],
|
|
)
|
|
def test_get_most_authorized_role(permissions, expected_role) -> None:
|
|
assert get_most_authorized_role(permissions) == expected_role
|
|
|
|
|
|
class TestRBACPermission:
|
|
def test_get_view_action(self) -> None:
|
|
viewset_action = "viewset_action"
|
|
viewset = MockedViewSet(viewset_action)
|
|
|
|
apiview = MockedAPIView()
|
|
|
|
method = "APIVIEW_ACTION"
|
|
request = MockedRequest(method=method)
|
|
|
|
assert RBACPermission._get_view_action(request, viewset) == viewset_action, "it works with a ViewSet"
|
|
assert RBACPermission._get_view_action(request, apiview) == method.lower(), "it works with an APIView"
|
|
|
|
def test_has_permission_works_on_a_viewset_view(self) -> None:
|
|
required_permission = RBACPermission.Permissions.ALERT_GROUPS_READ
|
|
|
|
action = "hello"
|
|
viewset = MockedViewSet(
|
|
action=action,
|
|
rbac_permissions={
|
|
action: [required_permission],
|
|
},
|
|
)
|
|
|
|
viewset_with_no_required_permissions = MockedViewSet(
|
|
action=action,
|
|
rbac_permissions={
|
|
action: [],
|
|
},
|
|
)
|
|
|
|
user_with_permission = MockedUser([required_permission])
|
|
user_without_permission = MockedUser([RBACPermission.Permissions.ALERT_GROUPS_WRITE])
|
|
|
|
assert (
|
|
RBACPermission().has_permission(MockedRequest(user_with_permission), viewset) is True
|
|
), "it works on a viewset when the user does have permission"
|
|
|
|
assert (
|
|
RBACPermission().has_permission(MockedRequest(user_without_permission), viewset) is False
|
|
), "it works on a viewset when the user does have permission"
|
|
|
|
assert (
|
|
RBACPermission().has_permission(
|
|
MockedRequest(user_without_permission), viewset_with_no_required_permissions
|
|
)
|
|
is True
|
|
), "it works on a viewset when the viewset action does not require permissions"
|
|
|
|
def test_has_permission_works_on_an_apiview_view(self) -> None:
|
|
required_permission = RBACPermission.Permissions.ALERT_GROUPS_READ
|
|
|
|
method = "hello"
|
|
apiview = MockedAPIView(
|
|
rbac_permissions={
|
|
method: [required_permission],
|
|
}
|
|
)
|
|
apiview_with_no_permissions = MockedAPIView(
|
|
rbac_permissions={
|
|
method: [],
|
|
}
|
|
)
|
|
|
|
user1 = MockedUser([required_permission])
|
|
user2 = MockedUser([RBACPermission.Permissions.ALERT_GROUPS_WRITE])
|
|
|
|
class Request(MockedRequest):
|
|
def __init__(self, user: typing.Optional[MockedUser] = None) -> None:
|
|
super().__init__(user, method)
|
|
|
|
assert (
|
|
RBACPermission().has_permission(Request(user1), apiview) is True
|
|
), "it works on an APIView when the user has permission"
|
|
|
|
assert (
|
|
RBACPermission().has_permission(Request(user2), apiview) is False
|
|
), "it works on an APIView when the user does not have permission"
|
|
|
|
assert (
|
|
RBACPermission().has_permission(Request(user2), apiview_with_no_permissions) is True
|
|
), "it works on a viewset when the viewset action does not require permissions"
|
|
|
|
def test_has_permission_throws_assertion_error_if_developer_forgets_to_specify_rbac_permissions(self) -> None:
|
|
action_slash_method = "hello"
|
|
error_msg = (
|
|
f"Must define a {RBAC_PERMISSIONS_ATTR} dict on the ViewSet that is consuming the RBACPermission class"
|
|
)
|
|
|
|
viewset = MockedViewSet(action_slash_method)
|
|
apiview = MockedAPIView()
|
|
|
|
with pytest.raises(AssertionError, match=error_msg):
|
|
RBACPermission().has_permission(MockedRequest(), viewset)
|
|
|
|
with pytest.raises(AssertionError, match=error_msg):
|
|
RBACPermission().has_permission(MockedRequest(method=action_slash_method), apiview)
|
|
|
|
def test_has_permission_throws_assertion_error_if_developer_forgets_to_specify_an_action_in_rbac_permissions(
|
|
self,
|
|
) -> None:
|
|
action_slash_method = "hello"
|
|
other_action_rbac_permissions = {"bonjour": []}
|
|
error_msg = f"""Each action must be defined within the {RBAC_PERMISSIONS_ATTR} dict on the ViewSet.
|
|
\nIf an action requires no permissions, its value should explicitly be set to an empty list"""
|
|
|
|
viewset = MockedViewSet(action_slash_method, other_action_rbac_permissions)
|
|
apiview = MockedAPIView(rbac_permissions=other_action_rbac_permissions)
|
|
|
|
with pytest.raises(AssertionError, match=error_msg):
|
|
RBACPermission().has_permission(MockedRequest(), viewset)
|
|
|
|
with pytest.raises(AssertionError, match=error_msg):
|
|
RBACPermission().has_permission(MockedRequest(method=action_slash_method), apiview)
|
|
|
|
def test_has_object_permission_returns_true_if_rbac_object_permissions_not_specified(self) -> None:
|
|
request = MockedRequest()
|
|
assert RBACPermission().has_object_permission(request, MockedAPIView(), None) is True
|
|
assert RBACPermission().has_object_permission(request, MockedViewSet("potato"), None) is True
|
|
|
|
def test_has_object_permission_works_if_no_permission_class_specified_for_action(self) -> None:
|
|
action = "hello"
|
|
|
|
request = MockedRequest(None, action)
|
|
apiview = MockedAPIView(rbac_object_permissions={})
|
|
viewset = MockedViewSet(action, rbac_object_permissions={})
|
|
|
|
assert RBACPermission().has_object_permission(request, apiview, None) is True
|
|
assert RBACPermission().has_object_permission(request, viewset, None) is True
|
|
|
|
def test_has_object_permission_works_when_permission_class_specified_for_action(self) -> None:
|
|
action = "hello"
|
|
mocked_permission_class_response = "asdfasdfasdf"
|
|
|
|
class MockedPermissionClass:
|
|
def has_object_permission(self, _req, _view, _obj) -> None:
|
|
return mocked_permission_class_response
|
|
|
|
rbac_object_permissions = {MockedPermissionClass(): (action,)}
|
|
request = MockedRequest(None, action)
|
|
apiview = MockedAPIView(rbac_object_permissions=rbac_object_permissions)
|
|
viewset = MockedViewSet(action, rbac_object_permissions=rbac_object_permissions)
|
|
|
|
assert RBACPermission().has_object_permission(request, apiview, None) == mocked_permission_class_response
|
|
assert RBACPermission().has_object_permission(request, viewset, None) == mocked_permission_class_response
|
|
|
|
|
|
class TestIsOwner:
|
|
def test_it_works_when_comparing_user_to_object(self) -> None:
|
|
user1 = MockedUser([])
|
|
user2 = MockedUser([])
|
|
request = MockedRequest(user1)
|
|
IsUser = IsOwner()
|
|
|
|
assert IsUser.has_object_permission(request, None, user1) is True
|
|
assert IsUser.has_object_permission(request, None, user2) is False
|
|
|
|
def test_it_works_when_comparing_user_to_ownership_field_object(self) -> None:
|
|
user1 = MockedUser([])
|
|
user2 = MockedUser([])
|
|
schedule = MockedSchedule(user1)
|
|
IsScheduleOwner = IsOwner("user")
|
|
|
|
assert IsScheduleOwner.has_object_permission(MockedRequest(user1), None, schedule) is True
|
|
assert IsScheduleOwner.has_object_permission(MockedRequest(user2), None, schedule) is False
|
|
|
|
def test_it_works_when_comparing_user_to_nested_ownership_field_object(self) -> None:
|
|
class Thingy:
|
|
def __init__(self, schedule: MockedSchedule) -> None:
|
|
self.schedule = schedule
|
|
|
|
user1 = MockedUser([])
|
|
user2 = MockedUser([])
|
|
schedule = MockedSchedule(user1)
|
|
thingy = Thingy(schedule)
|
|
IsScheduleOwner = IsOwner("schedule.user")
|
|
|
|
assert IsScheduleOwner.has_object_permission(MockedRequest(user1), None, thingy) is True
|
|
assert IsScheduleOwner.has_object_permission(MockedRequest(user2), None, thingy) is False
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"user_permissions,required_permissions,expected_result",
|
|
[
|
|
(
|
|
[RBACPermission.Permissions.ALERT_GROUPS_READ],
|
|
[RBACPermission.Permissions.ALERT_GROUPS_READ],
|
|
True,
|
|
),
|
|
(
|
|
[RBACPermission.Permissions.ALERT_GROUPS_READ, RBACPermission.Permissions.ALERT_GROUPS_WRITE],
|
|
[RBACPermission.Permissions.ALERT_GROUPS_READ, RBACPermission.Permissions.ALERT_GROUPS_WRITE],
|
|
True,
|
|
),
|
|
(
|
|
[RBACPermission.Permissions.ALERT_GROUPS_WRITE],
|
|
[RBACPermission.Permissions.ALERT_GROUPS_READ],
|
|
False,
|
|
),
|
|
(
|
|
[RBACPermission.Permissions.ALERT_GROUPS_READ],
|
|
[RBACPermission.Permissions.ALERT_GROUPS_READ, RBACPermission.Permissions.ALERT_GROUPS_WRITE],
|
|
False,
|
|
),
|
|
],
|
|
)
|
|
def test_HasRBACPermission(user_permissions, required_permissions, expected_result) -> None:
|
|
request = MockedRequest(MockedUser(user_permissions))
|
|
assert HasRBACPermissions(required_permissions).has_object_permission(request, None, None) == expected_result
|
|
|
|
|
|
class TestIsOwnerOrHasRBACPermissions:
|
|
required_permission = RBACPermission.Permissions.SCHEDULES_READ
|
|
required_permissions = [required_permission]
|
|
|
|
def test_it_works_when_user_is_owner_and_does_not_have_permissions(self) -> None:
|
|
user1 = MockedUser([])
|
|
schedule = MockedSchedule(user1)
|
|
request = MockedRequest(user1)
|
|
|
|
PermClass = IsOwnerOrHasRBACPermissions(self.required_permissions)
|
|
assert PermClass.has_object_permission(request, None, user1) is True
|
|
|
|
PermClass = IsOwnerOrHasRBACPermissions(self.required_permissions, "user")
|
|
assert PermClass.has_object_permission(request, None, schedule) is True
|
|
|
|
def test_it_works_when_user_is_owner_and_has_permissions(self) -> None:
|
|
user1 = MockedUser(self.required_permissions)
|
|
schedule = MockedSchedule(user1)
|
|
request = MockedRequest(user1)
|
|
|
|
PermClass = IsOwnerOrHasRBACPermissions(self.required_permissions)
|
|
assert PermClass.has_object_permission(request, None, user1) is True
|
|
|
|
PermClass = IsOwnerOrHasRBACPermissions(self.required_permissions, "user")
|
|
assert PermClass.has_object_permission(request, None, schedule) is True
|
|
|
|
def test_it_works_when_user_is_not_owner_and_does_not_have_permissions(self) -> None:
|
|
user1 = MockedUser([])
|
|
user2 = MockedUser([])
|
|
schedule = MockedSchedule(user1)
|
|
request = MockedRequest(user2)
|
|
|
|
PermClass = IsOwnerOrHasRBACPermissions(self.required_permissions)
|
|
assert PermClass.has_object_permission(request, None, user1) is False
|
|
|
|
PermClass = IsOwnerOrHasRBACPermissions(self.required_permissions, "user")
|
|
assert PermClass.has_object_permission(request, None, schedule) is False
|
|
|
|
def test_it_works_when_user_is_not_owner_and_has_permissions(self) -> None:
|
|
user1 = MockedUser([])
|
|
user2 = MockedUser(self.required_permissions)
|
|
schedule = MockedSchedule(user1)
|
|
request = MockedRequest(user2)
|
|
|
|
PermClass = IsOwnerOrHasRBACPermissions(self.required_permissions)
|
|
assert PermClass.has_object_permission(request, None, user1) is True
|
|
|
|
PermClass = IsOwnerOrHasRBACPermissions(self.required_permissions, "user")
|
|
assert PermClass.has_object_permission(request, None, schedule) is True
|
|
|
|
class Thingy:
|
|
def __init__(self, schedule: MockedSchedule) -> None:
|
|
self.schedule = schedule
|
|
|
|
thingy = Thingy(schedule)
|
|
PermClass = IsOwnerOrHasRBACPermissions(self.required_permissions, "schedule.user")
|
|
|
|
assert PermClass.has_object_permission(request, None, thingy) is True
|
|
assert PermClass.has_object_permission(MockedRequest(MockedUser([])), None, thingy) is False
|