diff --git a/accounts/migrations/0006_alter_userattributes_id_alter_userinstance_id_and_more.py b/accounts/migrations/0006_alter_userattributes_id_alter_userinstance_id_and_more.py new file mode 100644 index 00000000..401b6309 --- /dev/null +++ b/accounts/migrations/0006_alter_userattributes_id_alter_userinstance_id_and_more.py @@ -0,0 +1,41 @@ +# Generated by Django 4.2.30 on 2026-09-27 10:18 + +from django.conf import settings +from django.db import migrations, models +import django.db.models.deletion + + +class Migration(migrations.Migration): + + dependencies = [ + migrations.swappable_dependency(settings.AUTH_USER_MODEL), + ('accounts', '0005_auto_20200616_1039'), + ] + + operations = [ + migrations.AlterField( + model_name='userattributes', + name='id', + field=models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'), + ), + migrations.AlterField( + model_name='userinstance', + name='id', + field=models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'), + ), + migrations.AlterField( + model_name='usersshkey', + name='id', + field=models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'), + ), + migrations.AlterField( + model_name='usersshkey', + name='keypublic', + field=models.TextField(max_length=16384, verbose_name='public key'), + ), + migrations.AlterField( + model_name='usersshkey', + name='user', + field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL), + ), + ] diff --git a/accounts/models.py b/accounts/models.py index 88cbb232..189de683 100644 --- a/accounts/models.py +++ b/accounts/models.py @@ -30,9 +30,9 @@ class UserInstance(models.Model): class UserSSHKey(models.Model): - user = models.ForeignKey(User, on_delete=models.DO_NOTHING) + user = models.ForeignKey(User, on_delete=models.CASCADE) keyname = models.CharField(_("key name"), max_length=25) - keypublic = models.CharField(_("public key"), max_length=500) + keypublic = models.TextField(_("public key"), max_length=16384) def __str__(self): return self.keyname diff --git a/accounts/test_security_fixes.py b/accounts/test_security_fixes.py new file mode 100644 index 00000000..1c77d3a4 --- /dev/null +++ b/accounts/test_security_fixes.py @@ -0,0 +1,64 @@ +from django.contrib.auth import get_user_model +from django.test import TestCase +from django.urls import reverse + +from accounts.models import UserInstance, UserSSHKey +from accounts.utils import validate_ssh_key +from computes.models import Compute +from instances.models import Instance + + +class AccountsSecurityFixesTestCase(TestCase): + def setUp(self): + User = get_user_model() + self.admin = User.objects.create_superuser( + username="sec_acc_admin", email="admin@example.com", password="password" + ) + self.user = User.objects.create_user( + username="sec_acc_user", password="password" + ) + self.compute = Compute.objects.create( + name="acc-comp", hostname="127.0.0.1", login="root", type=1 + ) + self.instance = Instance.objects.create( + compute=self.compute, name="acc-vm", uuid="99999999-9999-9999-9999-999999999999" + ) + self.user_inst = UserInstance.objects.create( + instance=self.instance, user=self.user + ) + + def test_open_redirect_prevented_in_user_instance_delete(self): + self.client.force_login(self.admin) + url = reverse("accounts:user_instance_delete", args=[self.user_inst.id]) + + # Attempt external malicious redirect + res = self.client.post(f"{url}?next=https://malicious-attacker.com") + self.assertEqual(res.status_code, 302) + # Should redirect to safe internal URL (user account page), NOT external site + self.assertNotIn("malicious-attacker.com", res.url) + self.assertEqual(res.url, reverse("accounts:account", args=[self.user.id])) + + def test_ssh_key_cascade_deletion(self): + key = UserSSHKey.objects.create( + user=self.user, + keyname="my-key", + keypublic="ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIGabcdef123456 test@example.com", + ) + self.assertTrue(UserSSHKey.objects.filter(id=key.id).exists()) + + # When user is deleted, key should be deleted via CASCADE (not left orphan) + self.user.delete() + self.assertFalse(UserSSHKey.objects.filter(id=key.id).exists()) + + def test_validate_ssh_key_supports_keys_with_and_without_comment(self): + rsa_base = "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQC6OOdbfv27QVnSC6sKxGaHb6YFc+3gxCkyVR3cTSXE/n5BEGf8aOgBpepULWa1RZfxYHY14PlKULDygdXSdrrR2kNSwoKz/Oo4d+3EE92L7ocl1+djZbptzgWgtw1OseLwbFik+iKlIdqPsH+IUQvX7yV545ZQtAP8Qj1R+uCqkw==" + # Key with comment + key_with_comment = f"{rsa_base} test@test" + self.assertTrue(validate_ssh_key(key_with_comment)) + + # Key without comment + key_without_comment = rsa_base + self.assertTrue(validate_ssh_key(key_without_comment)) + + # Invalid key + self.assertFalse(validate_ssh_key("not a valid key string")) diff --git a/accounts/utils.py b/accounts/utils.py index fbbdf811..a07a39b0 100644 --- a/accounts/utils.py +++ b/accounts/utils.py @@ -20,10 +20,9 @@ def get_user_totp_device(user): def validate_ssh_key(key): - array = key.encode().split() - # Each rsa-ssh key has 3 different strings in it, first one being - # typeofkey second one being keystring third one being username . - if len(array) != 3: + array = key.strip().encode().split() + # An OpenSSH public key has 2 parts (type, key) or 3 parts (type, key, comment). + if len(array) not in (2, 3): return False typeofkey = array[0] string = array[1] diff --git a/accounts/views.py b/accounts/views.py index a7da98a5..387c8718 100644 --- a/accounts/views.py +++ b/accounts/views.py @@ -7,6 +7,7 @@ from django.contrib.auth.forms import PasswordChangeForm from django.shortcuts import get_object_or_404, redirect, render from django.urls import reverse from django.http import HttpResponseRedirect +from django.utils.http import url_has_allowed_host_and_scheme from django.utils.translation import gettext_lazy as _ from instances.models import Instance @@ -161,11 +162,14 @@ def user_instance_delete(request, pk): if request.method == "POST": user = user_instance.user user_instance.delete() - next = request.GET.get("next", None) - if next: - return redirect(next) - else: - return redirect(reverse("accounts:account", args=[user.id])) + next_url = request.GET.get("next", None) + if next_url and url_has_allowed_host_and_scheme( + url=next_url, + allowed_hosts={request.get_host()}, + require_https=request.is_secure(), + ): + return redirect(next_url) + return redirect(reverse("accounts:account", args=[user.id])) return render( request, diff --git a/instances/api/viewsets.py b/instances/api/viewsets.py index afa2deb9..cde1b9d2 100644 --- a/instances/api/viewsets.py +++ b/instances/api/viewsets.py @@ -2,6 +2,7 @@ from admin.permissions import IsSuperUser from appsettings.settings import app_settings from computes import utils from computes.models import Compute +from django.http import Http404 from django.shortcuts import get_object_or_404 from instances.models import Flavor, Instance from instances.utils import migrate_instance @@ -99,41 +100,64 @@ class InstanceViewSet(viewsets.ViewSet): def retrieve(self, request, pk=None, compute_pk=None): queryset = get_instance(request.user, pk) + if compute_pk is not None and str(queryset.compute_id) != str(compute_pk): + raise Http404() serializer = InstanceDetailsSerializer(queryset, context={"request": request}) return Response(serializer.data) def destroy(self, request, pk=None, compute_pk=None): + queryset = get_instance(request.user, pk, perm_type="delete") + if compute_pk is not None and str(queryset.compute_id) != str(compute_pk): + raise Http404() instance_destroy(request, pk) return Response({"status": "Instance is destroyed"}) @action(detail=True, methods=["post"]) - def poweron(self, request, pk=None): + def poweron(self, request, pk=None, compute_pk=None): + queryset = get_instance(request.user, pk, perm_type="power") + if compute_pk is not None and str(queryset.compute_id) != str(compute_pk): + raise Http404() poweron(request, pk) return Response({"status": "poweron command send"}) @action(detail=True, methods=["post"]) - def poweroff(self, request, pk=None): + def poweroff(self, request, pk=None, compute_pk=None): + queryset = get_instance(request.user, pk, perm_type="power") + if compute_pk is not None and str(queryset.compute_id) != str(compute_pk): + raise Http404() poweroff(request, pk) return Response({"status": "poweroff command send"}) @action(detail=True, methods=["post"]) - def powercycle(self, request, pk=None): + def powercycle(self, request, pk=None, compute_pk=None): + queryset = get_instance(request.user, pk, perm_type="power") + if compute_pk is not None and str(queryset.compute_id) != str(compute_pk): + raise Http404() powercycle(request, pk) return Response({"status": "powercycle command send"}) @action(detail=True, methods=["post"]) - def forceoff(self, request, pk=None): + def forceoff(self, request, pk=None, compute_pk=None): + queryset = get_instance(request.user, pk, perm_type="power") + if compute_pk is not None and str(queryset.compute_id) != str(compute_pk): + raise Http404() force_off(request, pk) return Response({"status": "force off command send"}) @action(detail=True, methods=["post"]) - def suspend(self, request, pk=None): + def suspend(self, request, pk=None, compute_pk=None): + queryset = get_instance(request.user, pk) + if compute_pk is not None and str(queryset.compute_id) != str(compute_pk): + raise Http404() suspend(request, pk) return Response({"status": "suspend command send"}) @action(detail=True, methods=["post"]) - def resume(self, request, pk=None): + def resume(self, request, pk=None, compute_pk=None): + queryset = get_instance(request.user, pk) + if compute_pk is not None and str(queryset.compute_id) != str(compute_pk): + raise Http404() resume(request, pk) return Response({"status": "resume command send"}) diff --git a/instances/templates/instance_actions.html b/instances/templates/instance_actions.html index ccdcfdae..f3f24fdc 100644 --- a/instances/templates/instance_actions.html +++ b/instances/templates/instance_actions.html @@ -1,48 +1,71 @@ {% load i18n %} {% load bootstrap_icons %} -