before review
This commit is contained in:
1 parent
c6a830cb92
commit
d9881bde01
40 files changed
+1560
-158
No files matched your search
@@ -1,15 +1,18 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import timedelta
|
||||
from types import SimpleNamespace
|
||||
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.contrib.auth.models import Permission
|
||||
from django.test import TestCase
|
||||
from django.urls import reverse
|
||||
from django.utils import timezone
|
||||
from guardian.shortcuts import assign_perm
|
||||
from ninja.errors import HttpError
|
||||
|
||||
from apps.access.models import AccessRequest
|
||||
from apps.accounts.models import ErasureRequest
|
||||
from apps.core import rbac
|
||||
from apps.servers.models import Server, ServerAccount
|
||||
from apps.servers.permissions import user_can_logs, user_can_shell, user_can_users, user_has_any_access
|
||||
@@ -109,6 +112,32 @@ class ServerAccessScopeTests(TestCase):
|
||||
).exists()
|
||||
)
|
||||
|
||||
def test_dashboard_request_access_scoped_options_are_saved(self) -> None:
|
||||
self._grant_perm(self.user, "access", "add_accessrequest")
|
||||
self.client.force_login(self.user)
|
||||
|
||||
response = self.client.post(
|
||||
reverse("servers:request_access", args=[self.server.id]),
|
||||
{
|
||||
"scoped_request": "1",
|
||||
"request_shell": "1",
|
||||
"request_logs": "1",
|
||||
"requested_duration_hours": "24",
|
||||
"requested_server_username": "ubuntu",
|
||||
"reason": "Investigate auth failure",
|
||||
},
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 302)
|
||||
access_request = AccessRequest.objects.get(requester=self.user, server=self.server)
|
||||
self.assertEqual(access_request.status, AccessRequest.Status.PENDING)
|
||||
self.assertTrue(access_request.request_shell)
|
||||
self.assertTrue(access_request.request_logs)
|
||||
self.assertFalse(access_request.request_users)
|
||||
self.assertEqual(access_request.requested_duration_hours, 24)
|
||||
self.assertEqual(access_request.requested_server_username, "ubuntu")
|
||||
self.assertEqual(access_request.reason, "Investigate auth failure")
|
||||
|
||||
def test_admin_can_approve_pending_request(self) -> None:
|
||||
pending = AccessRequest.objects.create(
|
||||
requester=self.user,
|
||||
@@ -130,6 +159,28 @@ class ServerAccessScopeTests(TestCase):
|
||||
self.assertEqual(pending.status, AccessRequest.Status.APPROVED)
|
||||
self.assertEqual(pending.decided_by_id, self.admin.id)
|
||||
|
||||
def test_admin_approve_uses_requested_duration_for_expiry(self) -> None:
|
||||
pending = AccessRequest.objects.create(
|
||||
requester=self.user,
|
||||
server=self.server,
|
||||
status=AccessRequest.Status.PENDING,
|
||||
request_shell=True,
|
||||
requested_duration_hours=4,
|
||||
)
|
||||
self.client.force_login(self.admin)
|
||||
|
||||
decision_started_at = timezone.now()
|
||||
response = self.client.post(
|
||||
reverse("servers:decide_access_request", args=[pending.id]),
|
||||
{"action": "approve"},
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 302)
|
||||
pending.refresh_from_db()
|
||||
self.assertEqual(pending.status, AccessRequest.Status.APPROVED)
|
||||
self.assertIsNotNone(pending.expires_at)
|
||||
self.assertGreaterEqual(pending.expires_at, decision_started_at + timedelta(hours=3, minutes=59))
|
||||
|
||||
def test_admin_dashboard_requires_admin(self) -> None:
|
||||
self.client.force_login(self.user)
|
||||
response = self.client.get(reverse("servers:admin_dashboard"))
|
||||
@@ -149,6 +200,104 @@ class ServerAccessScopeTests(TestCase):
|
||||
self.assertContains(response, pending.requester.username)
|
||||
self.assertContains(response, "Access request queue")
|
||||
|
||||
def test_admin_dashboard_users_view_shows_gdpr_marker(self) -> None:
|
||||
ErasureRequest.objects.create(
|
||||
user=self.user,
|
||||
reason="Please erase my profile data completely.",
|
||||
status=ErasureRequest.Status.PENDING,
|
||||
)
|
||||
self.client.force_login(self.admin)
|
||||
|
||||
response = self.client.get(reverse("servers:admin_dashboard"), {"view": "users"})
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertContains(response, "Send reset email")
|
||||
self.assertContains(response, "GDPR request")
|
||||
|
||||
def test_admin_delete_user_from_dashboard(self) -> None:
|
||||
User = get_user_model()
|
||||
victim = User.objects.create_user(
|
||||
username="victim",
|
||||
email="victim@example.com",
|
||||
password="pass12345",
|
||||
)
|
||||
self.client.force_login(self.admin)
|
||||
|
||||
response = self.client.post(
|
||||
reverse("servers:admin_delete_user", args=[victim.id]),
|
||||
{"next": reverse("servers:admin_dashboard")},
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 302)
|
||||
self.assertFalse(User.objects.filter(id=victim.id).exists())
|
||||
|
||||
def test_admin_send_password_reset_from_dashboard(self) -> None:
|
||||
self.client.force_login(self.admin)
|
||||
|
||||
response = self.client.post(
|
||||
reverse("servers:admin_send_password_reset", args=[self.user.id]),
|
||||
{"next": reverse("servers:admin_dashboard")},
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 302)
|
||||
|
||||
def test_admin_delete_server_from_dashboard(self) -> None:
|
||||
server = Server.objects.create(display_name="delete-me", hostname="delete-me.example.test")
|
||||
self.client.force_login(self.admin)
|
||||
|
||||
response = self.client.post(
|
||||
reverse("servers:admin_delete_server", args=[server.id]),
|
||||
{"next": reverse("servers:admin_dashboard")},
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 302)
|
||||
self.assertFalse(Server.objects.filter(id=server.id).exists())
|
||||
|
||||
def test_admin_revoke_grant_from_dashboard(self) -> None:
|
||||
grant = AccessRequest.objects.create(
|
||||
requester=self.user,
|
||||
server=self.server,
|
||||
status=AccessRequest.Status.APPROVED,
|
||||
request_shell=True,
|
||||
expires_at=timezone.now() + timedelta(days=3),
|
||||
)
|
||||
self.client.force_login(self.admin)
|
||||
|
||||
response = self.client.post(
|
||||
reverse("servers:admin_revoke_grant", args=[grant.id]),
|
||||
{"next": reverse("servers:admin_dashboard")},
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 302)
|
||||
grant.refresh_from_db()
|
||||
self.assertEqual(grant.status, AccessRequest.Status.REVOKED)
|
||||
self.assertEqual(grant.decided_by_id, self.admin.id)
|
||||
self.assertIsNotNone(grant.expires_at)
|
||||
|
||||
def test_admin_change_grant_expiry_from_dashboard(self) -> None:
|
||||
grant = AccessRequest.objects.create(
|
||||
requester=self.user,
|
||||
server=self.server,
|
||||
status=AccessRequest.Status.APPROVED,
|
||||
request_shell=True,
|
||||
expires_at=timezone.now() + timedelta(days=1),
|
||||
)
|
||||
new_expiry = timezone.now() + timedelta(days=9)
|
||||
self.client.force_login(self.admin)
|
||||
|
||||
response = self.client.post(
|
||||
reverse("servers:admin_change_grant_expiry", args=[grant.id]),
|
||||
{
|
||||
"expires_at": timezone.localtime(new_expiry).strftime("%Y-%m-%dT%H:%M"),
|
||||
"next": reverse("servers:admin_dashboard"),
|
||||
},
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 302)
|
||||
grant.refresh_from_db()
|
||||
self.assertIsNotNone(grant.expires_at)
|
||||
self.assertGreater(grant.expires_at, timezone.now() + timedelta(days=8))
|
||||
|
||||
def test_server_admin_requires_admin(self) -> None:
|
||||
assign_perm("view_server", self.user, self.server)
|
||||
self.client.force_login(self.user)
|
||||
|
||||
Reference in new issue
Block a user