mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-08 11:16:47 +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.
99 lines
4.3 KiB
Python
99 lines
4.3 KiB
Python
import json
|
|
from unittest.mock import patch
|
|
|
|
from django.contrib.auth import get_user_model
|
|
from django.test import RequestFactory, TestCase
|
|
from django.urls import reverse
|
|
|
|
from accounts.models import UserInstance, UserSSHKey
|
|
from computes.models import Compute
|
|
from instances.models import Instance
|
|
from datasource.views import get_client_ip, get_hostname_by_ip
|
|
|
|
|
|
class DataSourceTestCase(TestCase):
|
|
def setUp(self):
|
|
self.factory = RequestFactory()
|
|
self.client.login(request=RequestFactory().get("/"), username="admin", password="admin")
|
|
self.compute = Compute.objects.create(
|
|
name="test_compute",
|
|
hostname="localhost",
|
|
type=1,
|
|
)
|
|
|
|
def test_os_index(self):
|
|
response = self.client.get(reverse("ds_openstack_index"))
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn("latest", response.content.decode("utf-8"))
|
|
|
|
def test_os_metadata_json_latest(self):
|
|
response = self.client.get(reverse("ds_openstack_metadata", args=["latest"]))
|
|
self.assertEqual(response.status_code, 200)
|
|
data = json.loads(response.content.decode("utf-8"))
|
|
self.assertEqual(data["uuid"], "iid-dswebvirtcloud")
|
|
self.assertIn("hostname", data)
|
|
|
|
def test_os_metadata_json_invalid_version(self):
|
|
response = self.client.get(reverse("ds_openstack_metadata", args=["v1"]))
|
|
self.assertEqual(response.status_code, 404)
|
|
|
|
def test_os_userdata_invalid_version(self):
|
|
response = self.client.get(reverse("ds_openstack_userdata", args=["v1"]))
|
|
self.assertEqual(response.status_code, 404)
|
|
|
|
def test_get_client_ip_direct(self):
|
|
request = self.factory.get("/")
|
|
request.META["REMOTE_ADDR"] = "192.168.1.10"
|
|
self.assertEqual(get_client_ip(request), "192.168.1.10")
|
|
|
|
def test_get_client_ip_forwarded(self):
|
|
request = self.factory.get("/")
|
|
request.META["HTTP_X_FORWARDED_FOR"] = "10.0.0.1, 192.168.1.50"
|
|
self.assertEqual(get_client_ip(request), "192.168.1.50")
|
|
|
|
def test_get_hostname_by_ip_fallback(self):
|
|
# Invalid IP will cause gethostbyaddr to fail and return the IP string itself
|
|
self.assertEqual(get_hostname_by_ip("256.256.256.256"), "256.256.256.256")
|
|
|
|
def test_vdi_url_not_found_compute(self):
|
|
response = self.client.get(reverse("vdi_url", args=[9999, "vm1"]))
|
|
self.assertEqual(response.status_code, 404)
|
|
|
|
|
|
class CloudInitTestCase(TestCase):
|
|
"""VMs fetch the cloud-init endpoints without logging in; keys go only to
|
|
a VM whose name is unique."""
|
|
|
|
KEY = "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIOwner owner@example"
|
|
|
|
def setUp(self):
|
|
self.compute = Compute.objects.create(name="ci-a", hostname="a.example", login="", password="", type=4)
|
|
self.owner = get_user_model().objects.create_user("ci_owner", password="pw")
|
|
UserSSHKey.objects.create(user=self.owner, keyname="k", keypublic=self.KEY)
|
|
self.vm = Instance.objects.create(compute=self.compute, name="web1", uuid="11111111-2222-3333-4444-999999999991")
|
|
UserInstance.objects.create(instance=self.vm, user=self.owner)
|
|
patcher = patch("datasource.views.get_hostname_by_ip", return_value="web1.lan")
|
|
patcher.start()
|
|
self.addCleanup(patcher.stop)
|
|
|
|
def userdata(self):
|
|
response = self.client.get(reverse("ds_openstack_userdata", args=["latest"]))
|
|
self.assertEqual(response.status_code, 200)
|
|
return response.content.decode()
|
|
|
|
def test_endpoints_need_no_login(self):
|
|
for name, args in (("ds_openstack_index", []), ("ds_openstack_metadata", ["latest"])):
|
|
self.assertEqual(self.client.get(reverse(name, args=args)).status_code, 200, name)
|
|
self.assertEqual(json.loads(self.client.get(reverse("ds_openstack_metadata", args=["latest"])).content)["hostname"], "web1.lan")
|
|
|
|
def test_owner_keys_for_a_unique_name(self):
|
|
self.assertIn(self.KEY, self.userdata())
|
|
|
|
def test_no_keys_when_the_name_is_taken_twice(self):
|
|
other = Compute.objects.create(name="ci-b", hostname="b.example", login="", password="", type=4)
|
|
Instance.objects.create(compute=other, name="web1", uuid="11111111-2222-3333-4444-999999999992")
|
|
self.assertNotIn("ssh", self.userdata())
|
|
|
|
def test_no_keys_for_an_unknown_name(self):
|
|
self.vm.delete()
|
|
self.assertNotIn("ssh", self.userdata())
|