mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-08 11:16:47 +00:00
51dedb7b8f
Disk edits: - edit_disk built a new <disk> from the form and replaced the old one, losing what the form does not show (e.g. encryption and virtio driver options); a bus change detached and re-attached the disk. The disk of the persistent definition is now edited in place and the domain defined once; a bus change picks a free device name and drops the old address and virtio-only options. Pending persistent changes of a running VM are kept. - Clone omits empty disk devices, refuses a disk without a supported source and destination, and checks the name before allocating storage. Locks: - Every VM change held its compute's exclusive lock, so one slow operation blocked all VMs on that compute. A VM change now holds its own VM lock plus a shared compute lock; reconciliation, migration and cloning hold the compute exclusively (two clones of different VMs must not pass the name check together). - Quota-checked changes of one user (CPU, memory and disk resize, clone) run one at a time under a per-user lock, also across computes, so concurrent requests cannot all pass the check against the old allocation. It is the outermost lock; superusers, who have no quota, skip it. - A busy lock answers 409 to the API and a message in the web UI. API power and destroy actions return the view's error instead of success.
189 lines
7.5 KiB
Python
189 lines
7.5 KiB
Python
from unittest.mock import patch
|
|
from accounts.models import UserInstance
|
|
from computes.api.serializers import ComputeSerializer
|
|
from computes.models import Compute
|
|
from django.contrib.auth import get_user_model
|
|
from django.test import TestCase
|
|
from instances.models import Flavor, Instance
|
|
from rest_framework.test import APIClient
|
|
|
|
User = get_user_model()
|
|
|
|
|
|
class APISecurityTestCase(TestCase):
|
|
def setUp(self):
|
|
self.client = APIClient()
|
|
self.admin = User.objects.create_superuser(
|
|
username="admin_api_test",
|
|
email="admin_api@example.com",
|
|
password="adminpassword",
|
|
)
|
|
self.regular_user = User.objects.create_user(
|
|
username="regular_api_user",
|
|
email="user@example.com",
|
|
password="userpassword",
|
|
)
|
|
self.staff_user = User.objects.create_user(
|
|
username="staff_api_user",
|
|
email="staff@example.com",
|
|
password="staffpassword",
|
|
is_staff=True,
|
|
is_superuser=False,
|
|
)
|
|
self.compute = Compute.objects.create(
|
|
name="api-test-compute",
|
|
hostname="127.0.0.1",
|
|
login="libvirt_user",
|
|
password="secret_hypervisor_password",
|
|
type=1,
|
|
)
|
|
self.empty_compute = Compute.objects.create(
|
|
name="empty-test-compute",
|
|
hostname="127.0.0.1",
|
|
login="libvirt_user",
|
|
password="secret_hypervisor_password",
|
|
type=1,
|
|
)
|
|
self.flavor = Flavor.objects.create(
|
|
label="test-flavor",
|
|
vcpu=2,
|
|
memory=2048,
|
|
disk=20,
|
|
)
|
|
self.instance1 = Instance.objects.create(
|
|
compute=self.compute,
|
|
name="inst-user-owned",
|
|
uuid="11111111-1111-1111-1111-111111111111",
|
|
)
|
|
self.instance2 = Instance.objects.create(
|
|
compute=self.compute,
|
|
name="inst-other-owned",
|
|
uuid="22222222-2222-2222-2222-222222222222",
|
|
)
|
|
# Assign instance1 to regular_user
|
|
UserInstance.objects.create(user=self.regular_user, instance=self.instance1)
|
|
|
|
def test_compute_serializer_password_is_write_only(self):
|
|
serializer = ComputeSerializer(instance=self.compute)
|
|
self.assertNotIn("password", serializer.data)
|
|
self.assertEqual(serializer.data["name"], "api-test-compute")
|
|
self.assertEqual(serializer.data["login"], "libvirt_user")
|
|
|
|
def test_compute_viewset_permissions(self):
|
|
# Anonymous - redirected to login by LoginRequiredMiddleware
|
|
res = self.client.get("/api/v1/computes/")
|
|
self.assertIn(res.status_code, [302, 401, 403])
|
|
|
|
# Regular user - forbidden by IsSuperUser
|
|
self.client.force_login(self.regular_user)
|
|
res = self.client.get("/api/v1/computes/")
|
|
self.assertEqual(res.status_code, 403)
|
|
|
|
# Staff user (not superuser) - forbidden by IsSuperUser
|
|
self.client.force_login(self.staff_user)
|
|
res = self.client.get("/api/v1/computes/")
|
|
self.assertEqual(res.status_code, 403)
|
|
|
|
# Admin user - allowed
|
|
self.client.force_login(self.admin)
|
|
res = self.client.get("/api/v1/computes/")
|
|
self.assertEqual(res.status_code, 200)
|
|
|
|
@patch("computes.utils.refresh_instance_database")
|
|
def test_compute_instance_list_isolation(self, mock_refresh):
|
|
# Regular user should only see instance1 on compute
|
|
self.client.force_login(self.regular_user)
|
|
res = self.client.get(f"/api/v1/computes/{self.compute.id}/instances/")
|
|
self.assertEqual(res.status_code, 200)
|
|
names = [item["name"] for item in res.data]
|
|
self.assertIn("inst-user-owned", names)
|
|
self.assertNotIn("inst-other-owned", names)
|
|
mock_refresh.assert_called_once_with(self.compute)
|
|
|
|
# Regular user accessing compute where they have NO instances:
|
|
# returns empty list without calling refresh_instance_database!
|
|
mock_refresh.reset_mock()
|
|
res_empty = self.client.get(f"/api/v1/computes/{self.empty_compute.id}/instances/")
|
|
self.assertEqual(res_empty.status_code, 200)
|
|
self.assertEqual(res_empty.data, [])
|
|
mock_refresh.assert_not_called()
|
|
|
|
# Admin user should see both instances
|
|
self.client.force_login(self.admin)
|
|
res = self.client.get(f"/api/v1/computes/{self.compute.id}/instances/")
|
|
self.assertEqual(res.status_code, 200)
|
|
names = [item["name"] for item in res.data]
|
|
self.assertIn("inst-user-owned", names)
|
|
self.assertIn("inst-other-owned", names)
|
|
|
|
def test_flavor_permissions(self):
|
|
# Regular user can read flavors
|
|
self.client.force_login(self.regular_user)
|
|
res = self.client.get("/api/v1/flavor/")
|
|
self.assertEqual(res.status_code, 200)
|
|
|
|
# Regular user cannot mutate flavors
|
|
res = self.client.post(
|
|
"/api/v1/flavor/",
|
|
{"label": "malicious", "vcpu": 4, "memory": 4096, "disk": 40},
|
|
)
|
|
self.assertEqual(res.status_code, 403)
|
|
|
|
# Staff user (not superuser) cannot mutate flavors
|
|
self.client.force_login(self.staff_user)
|
|
res = self.client.post(
|
|
"/api/v1/flavor/",
|
|
{"label": "staffmutate", "vcpu": 4, "memory": 4096, "disk": 40},
|
|
)
|
|
self.assertEqual(res.status_code, 403)
|
|
|
|
res = self.client.delete(f"/api/v1/flavor/{self.flavor.id}/")
|
|
self.assertEqual(res.status_code, 403)
|
|
|
|
def test_migrate_viewset_permissions(self):
|
|
self.client.force_login(self.regular_user)
|
|
res = self.client.post("/api/v1/migrate/", {})
|
|
self.assertEqual(res.status_code, 403)
|
|
|
|
self.client.force_login(self.staff_user)
|
|
res = self.client.post("/api/v1/migrate/", {})
|
|
self.assertEqual(res.status_code, 403)
|
|
|
|
def test_network_viewset_permissions(self):
|
|
self.client.force_login(self.regular_user)
|
|
res = self.client.get(f"/api/v1/computes/{self.compute.id}/networks/")
|
|
self.assertEqual(res.status_code, 403)
|
|
|
|
self.client.force_login(self.staff_user)
|
|
res = self.client.get(f"/api/v1/computes/{self.compute.id}/networks/")
|
|
self.assertEqual(res.status_code, 403)
|
|
|
|
def test_interface_viewset_permissions(self):
|
|
self.client.force_login(self.regular_user)
|
|
res = self.client.get(f"/api/v1/computes/{self.compute.id}/interfaces/")
|
|
self.assertEqual(res.status_code, 403)
|
|
|
|
self.client.force_login(self.staff_user)
|
|
res = self.client.get(f"/api/v1/computes/{self.compute.id}/interfaces/")
|
|
self.assertEqual(res.status_code, 403)
|
|
|
|
def test_storage_viewset_permissions(self):
|
|
self.client.force_login(self.regular_user)
|
|
res = self.client.get(f"/api/v1/computes/{self.compute.id}/storages/")
|
|
self.assertEqual(res.status_code, 403)
|
|
|
|
self.client.force_login(self.staff_user)
|
|
res = self.client.get(f"/api/v1/computes/{self.compute.id}/storages/")
|
|
self.assertEqual(res.status_code, 403)
|
|
|
|
def test_busy_instance_api_does_not_report_success(self):
|
|
self.client.force_login(self.regular_user)
|
|
with patch("instances.views.libvirt_instance_lock", side_effect=TimeoutError("busy")), patch(
|
|
"instances.models.wvmInstance"
|
|
) as proxy:
|
|
response = self.client.post(
|
|
f"/api/v1/computes/{self.compute.pk}/instances/{self.instance1.pk}/poweron/", {}
|
|
)
|
|
self.assertEqual(response.status_code, 409)
|
|
self.assertIn("busy", response.json()["detail"])
|
|
proxy.assert_not_called()
|