mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-08 19:26:48 +00:00
c2fea2aa39
Nothing limited failed logins: a password could be guessed at the speed the server answers. django-axes (8.3.1, Django 5.2, Python 3.10-3.13) now counts them. - A user is locked for the address the failed logins come from (username and IP together), so an attacker cannot lock someone out from elsewhere and users behind one NAT address do not share a lock. A successful login resets the count. axes sits in front of every authentication: the login page with or without OTP, LDAP and the API. - The number of failed logins (default 5) and the lock time in minutes (default 15) are on the Settings page (LOGIN_FAILURE_LIMIT, LOGIN_LOCKOUT_MINUTES, added by migration appsettings.0014) and apply at once; an invalid value falls back to the default. Unlock with "manage.py axes_reset_username <username>". - Behind the bundled nginx every request comes from 127.0.0.1; the client address is read from X-Real-IP, which nginx sets, but only for requests from a loopback address or from LOGIN_TRUSTED_PROXIES, so a forged header does not escape a lock. settings.py.template gets the app, middleware, backend and AXES_* settings; README tells existing installations what to copy. Tests that call client.login now pass a request, which the axes backend needs. The new tests fail with axes disabled.
113 lines
4.7 KiB
Python
113 lines
4.7 KiB
Python
import json
|
|
import socket
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from django.test import RequestFactory, TestCase
|
|
from django.urls import reverse
|
|
|
|
from computes.models import Compute
|
|
from instances.models import Instance
|
|
|
|
|
|
class SetRootPassTestCase(TestCase):
|
|
def setUp(self):
|
|
self.client.login(request=RequestFactory().get("/"), username="admin", password="admin")
|
|
self.compute = Compute.objects.create(
|
|
name="test_compute_root_pass",
|
|
hostname="localhost",
|
|
type=1,
|
|
)
|
|
self.instance = Instance.objects.create(
|
|
compute=self.compute,
|
|
name="root-pass-vm",
|
|
uuid="87654321-4321-4321-4321-210987654321",
|
|
)
|
|
|
|
@patch("socket.socket")
|
|
@patch.object(Instance, "proxy")
|
|
def test_set_root_pass_post(self, mock_proxy, mock_socket_cls):
|
|
# Renamed on the host: gstfsd must get the name of this UUID's domain
|
|
mock_proxy.instance.name.return_value = "renamed-on-host"
|
|
mock_proxy.get_status.return_value = 5 # status: running
|
|
mock_sock = MagicMock()
|
|
mock_socket_cls.return_value = mock_sock
|
|
mock_sock.recv.return_value = json.dumps({"return": "success"}).encode()
|
|
|
|
response = self.client.post(
|
|
reverse("instances:rootpasswd", args=[self.instance.id]),
|
|
{"passwd": "newSecretPassword123"}
|
|
)
|
|
# Redirects back to referer or instance detail
|
|
self.assertEqual(response.status_code, 302)
|
|
|
|
# Check socket sent the command with hashed password
|
|
self.assertTrue(mock_sock.send.called)
|
|
sent_bytes = mock_sock.send.call_args[0][0]
|
|
data = json.loads(sent_bytes.decode())
|
|
self.assertEqual(data["action"], "password")
|
|
self.assertEqual(data["vname"], "renamed-on-host")
|
|
# SHA-512 crypt with a random salt, not the old fixed one
|
|
self.assertTrue(data["passwd"].startswith("$6$"))
|
|
self.assertFalse(data["passwd"].startswith("$6$kgPoiREy$"))
|
|
|
|
def test_set_root_pass_get_not_allowed(self):
|
|
response = self.client.get(reverse("instances:rootpasswd", args=[self.instance.id]))
|
|
self.assertEqual(response.status_code, 405)
|
|
|
|
def test_set_root_pass_not_found(self):
|
|
response = self.client.post(reverse("instances:rootpasswd", args=[99999]), {"passwd": "secret"})
|
|
self.assertEqual(response.status_code, 404)
|
|
|
|
@patch("socket.socket")
|
|
@patch.object(Instance, "proxy")
|
|
def test_same_password_gets_a_different_salt_each_time(self, mock_proxy, mock_socket_cls):
|
|
mock_proxy.instance.name.return_value = "root-pass-vm"
|
|
mock_proxy.get_status.return_value = 5
|
|
mock_sock = MagicMock()
|
|
mock_socket_cls.return_value = mock_sock
|
|
mock_sock.recv.return_value = json.dumps({"return": "success"}).encode()
|
|
url = reverse("instances:rootpasswd", args=[self.instance.id])
|
|
|
|
self.client.post(url, {"passwd": "samePassword1"})
|
|
self.client.post(url, {"passwd": "samePassword1"})
|
|
|
|
hashes = [json.loads(c[0][0].decode())["passwd"] for c in mock_sock.send.call_args_list]
|
|
self.assertEqual(len(hashes), 2)
|
|
self.assertNotEqual(hashes[0], hashes[1])
|
|
|
|
@patch("socket.socket")
|
|
@patch.object(Instance, "proxy")
|
|
def test_unreachable_gstfsd_is_an_error_message_not_a_crash(self, mock_proxy, mock_socket_cls):
|
|
mock_proxy.instance.name.return_value = "root-pass-vm"
|
|
mock_proxy.get_status.return_value = 5
|
|
mock_sock = MagicMock()
|
|
mock_socket_cls.return_value = mock_sock
|
|
mock_sock.connect.side_effect = socket.timeout("timed out")
|
|
|
|
response = self.client.post(
|
|
reverse("instances:rootpasswd", args=[self.instance.id]), {"passwd": "x"}
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 302)
|
|
mock_sock.settimeout.assert_called()
|
|
|
|
@patch("socket.socket")
|
|
@patch.object(Instance, "proxy")
|
|
def test_add_public_key_unreachable_gstfsd_is_an_error_message(self, mock_proxy, mock_socket_cls):
|
|
from accounts.models import UserSSHKey
|
|
from django.contrib.auth import get_user_model
|
|
|
|
admin = get_user_model().objects.get(username="admin")
|
|
key = UserSSHKey.objects.create(user=admin, keyname="k", keypublic="ssh-ed25519 AAAA k")
|
|
mock_proxy.instance.name.return_value = "root-pass-vm"
|
|
mock_proxy.get_status.return_value = 5
|
|
mock_sock = MagicMock()
|
|
mock_socket_cls.return_value = mock_sock
|
|
mock_sock.connect.side_effect = ConnectionRefusedError()
|
|
|
|
response = self.client.post(
|
|
reverse("instances:add_public_key", args=[self.instance.id]), {"sshkeyid": key.id}
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 302)
|
|
mock_sock.settimeout.assert_called()
|