1
0
Fork 0
mirror of https://github.com/retspen/webvirtcloud synced 2026-11-08 19:26:48 +00:00

feat(accounts): failed logins lock a user for the address they come from

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.
This commit is contained in:
catborise 2026-10-06 11:05:03 +03:00
parent 62ffaab808
commit c2fea2aa39
17 changed files with 223 additions and 27 deletions

View file

@ -413,6 +413,8 @@ to `MIDDLEWARE` in your existing `webvirtcloud/settings.py`, after
middleware, if present). A system check rejects a missing or misplaced middleware.
The template is not copied over an existing settings file during an upgrade.
Failed logins are limited with `django-axes` (installed from `conf/requirements.txt`). Copy the `axes` entries of `webvirtcloud/settings.py.template` into your `webvirtcloud/settings.py`: `"axes"` in `INSTALLED_APPS`, `"axes.middleware.AxesMiddleware"` at the end of `MIDDLEWARE`, `"axes.backends.AxesStandaloneBackend"` first in `AUTHENTICATION_BACKENDS`, and the `AXES_*` settings; then run `migrate`. The number of failed logins and the lock time are on the Settings page; unlock a user with `python3 manage.py axes_reset_username <username>`. Behind a proxy other than the bundled nginx, the proxy must set `X-Real-IP` to the client address and its address must be listed in `LOGIN_TRUSTED_PROXIES`.
Migration `accounts.0007` transfers the first-login requirement from any existing
`data/admin_password` file to the user record. Run migrations as a user who can read
that file. Thereafter, deleting or losing access to the file does not bypass the

54
accounts/lockout.py Normal file
View file

@ -0,0 +1,54 @@
"""
Settings for django-axes, read from the Settings page (AppSettings) on each
use, so an administrator can change them without a restart.
"""
from datetime import timedelta
from ipaddress import ip_address, ip_network
from appsettings.models import AppSettings
from django.conf import settings
DEFAULT_FAILURE_LIMIT = 5
DEFAULT_LOCKOUT_MINUTES = 15
def _positive_int(key, default):
try:
value = int(AppSettings.objects.get(key=key).value)
except (AppSettings.DoesNotExist, ValueError):
return default
return value if value > 0 else default
def failure_limit(request, credentials):
"""AXES_FAILURE_LIMIT: failed logins allowed before the lock."""
return _positive_int("LOGIN_FAILURE_LIMIT", DEFAULT_FAILURE_LIMIT)
def cooloff_time(request):
"""AXES_COOLOFF_TIME: how long a lock lasts."""
return timedelta(minutes=_positive_int("LOGIN_LOCKOUT_MINUTES", DEFAULT_LOCKOUT_MINUTES))
def _from_trusted_proxy(remote):
try:
address = ip_address(remote)
except ValueError:
return False
if address.is_loopback: # the bundled nginx
return True
return any(address in ip_network(net, strict=False) for net in getattr(settings, "LOGIN_TRUSTED_PROXIES", []))
def client_ip(request):
"""
AXES_CLIENT_IP_CALLABLE. Behind the bundled nginx every request comes from
127.0.0.1 and nginx sets X-Real-IP to the client's address (overwriting
any sent by the client). The header is used only for requests from that
proxy, or from the addresses in LOGIN_TRUSTED_PROXIES; from anyone else
it could be forged to escape a lock.
"""
remote = request.META.get("REMOTE_ADDR", "")
real_ip = request.META.get("HTTP_X_REAL_IP", "").strip()
return real_ip if real_ip and _from_trusted_proxy(remote) else remote

View file

@ -9,7 +9,7 @@ from unittest.mock import MagicMock, patch
from django.apps import apps
from django.contrib.auth import authenticate, get_user_model
from django.test import TestCase, override_settings
from django.test import RequestFactory, TestCase, override_settings
from accounts.apps import create_admin
from accounts.models import UserAttributes
@ -41,7 +41,7 @@ class GeneratedAdminPasswordTestCase(TestCase):
self.assertEqual(stat.S_IMODE(password_file.stat().st_mode), 0o600)
self.assertNotIn(password, output)
self.assertIn(str(password_file), output)
self.assertIsNotNone(authenticate(username="admin", password=password))
self.assertIsNotNone(authenticate(RequestFactory().get("/"), username="admin", password=password))
self.assertTrue(UserAttributes.objects.get(user__username="admin").must_change_password)
def test_does_not_follow_a_planted_symlink(self):

View file

@ -3,7 +3,7 @@ import tempfile
from pathlib import Path
from django.contrib.auth import get_user_model
from django.test import TestCase, override_settings
from django.test import RequestFactory, TestCase, override_settings
from django.urls import reverse
from unittest.mock import patch
@ -29,7 +29,7 @@ class FirstLoginPasswordChangeTestCase(TestCase):
UserAttributes.objects.create(user=self.admin, must_change_password=True)
def test_login_with_generated_password_redirects_every_page_to_change_form(self):
self.client.login(username="firstadmin", password=GENERATED)
self.client.login(request=RequestFactory().get("/"), username="firstadmin", password=GENERATED)
for url in (reverse("index"), reverse("instances:index"), reverse("accounts:profile")):
with self.subTest(url=url):
response = self.client.get(url)
@ -39,7 +39,7 @@ class FirstLoginPasswordChangeTestCase(TestCase):
self.assertEqual(self.client.get(reverse("accounts:change_password")).status_code, 200)
def test_changing_the_password_lifts_the_redirect_and_deletes_the_file(self):
self.client.login(username="firstadmin", password=GENERATED)
self.client.login(request=RequestFactory().get("/"), username="firstadmin", password=GENERATED)
self.client.post(
reverse("accounts:change_password"),
{
@ -57,7 +57,7 @@ class FirstLoginPasswordChangeTestCase(TestCase):
get_user_model().objects.create_superuser(
username="otheradmin", email="otheradmin@example.com", password="Other-pass-789"
)
self.client.login(username="otheradmin", password="Other-pass-789")
self.client.login(request=RequestFactory().get("/"), username="otheradmin", password="Other-pass-789")
self.assertEqual(self.client.get(reverse("instances:index")).status_code, 200)
self.assertTrue(self.password_file.exists())
@ -71,7 +71,7 @@ class FirstLoginPasswordChangeTestCase(TestCase):
def test_unreadable_password_file_is_not_used_during_login(self):
with patch("accounts.apps.admin_password_path", side_effect=AssertionError("must not read file")):
self.client.login(username="firstadmin", password=GENERATED)
self.client.login(request=RequestFactory().get("/"), username="firstadmin", password=GENERATED)
self.assertRedirects(
self.client.get(reverse("accounts:profile")),
reverse("accounts:change_password"), fetch_redirect_response=False,

View file

@ -0,0 +1,91 @@
"""Failed logins lock a username for the address it is tried from
(django-axes). The number of attempts and the lock time come from the
Settings page."""
from datetime import timedelta
from appsettings.models import AppSettings
from django.contrib.auth import get_user_model
from django.test import RequestFactory, TestCase, override_settings
from django.urls import reverse
from accounts.lockout import client_ip, cooloff_time, failure_limit
@override_settings(OTP_ENABLED=False)
class LoginLockoutTests(TestCase):
def setUp(self):
get_user_model().objects.create_user("lo-user", password="right-pass-1")
AppSettings.objects.filter(key="LOGIN_FAILURE_LIMIT").update(value="3")
def login(self, password, ip="203.0.113.10", via_proxy=True):
meta = {"REMOTE_ADDR": "127.0.0.1", "HTTP_X_REAL_IP": ip} if via_proxy else {"REMOTE_ADDR": ip}
return self.client.post(reverse("accounts:login"), {"username": "lo-user", "password": password}, **meta)
def test_the_limit_from_the_settings_page_locks_the_user_at_that_address(self):
for _ in range(3):
self.login("wrong")
self.assertEqual(self.login("right-pass-1").status_code, 429)
# another address is not locked: an attacker cannot lock someone out
self.assertEqual(self.login("right-pass-1", ip="198.51.100.7").status_code, 302)
def test_a_successful_login_starts_the_count_again(self):
self.login("wrong")
self.login("wrong")
self.assertEqual(self.login("right-pass-1").status_code, 302)
self.client.logout()
self.login("wrong")
self.login("wrong")
self.assertEqual(self.login("right-pass-1").status_code, 302)
def test_the_proxy_header_is_trusted_only_from_the_local_proxy(self):
# direct requests: X-Real-IP is ignored, so changing it does not escape the lock
for n in range(3):
self.client.post(reverse("accounts:login"), {"username": "lo-user", "password": "wrong"},
REMOTE_ADDR="203.0.113.20", HTTP_X_REAL_IP=f"10.0.0.{n}")
response = self.client.post(reverse("accounts:login"), {"username": "lo-user", "password": "right-pass-1"},
REMOTE_ADDR="203.0.113.20", HTTP_X_REAL_IP="10.0.0.99")
self.assertEqual(response.status_code, 429)
class LockoutSettingsTests(TestCase):
def request(self, remote, real_ip=None):
meta = {"REMOTE_ADDR": remote}
if real_ip:
meta["HTTP_X_REAL_IP"] = real_ip
return RequestFactory().post("/", **meta)
def test_values_from_the_settings_page(self):
AppSettings.objects.filter(key="LOGIN_FAILURE_LIMIT").update(value="7")
AppSettings.objects.filter(key="LOGIN_LOCKOUT_MINUTES").update(value="30")
self.assertEqual(failure_limit(None, {}), 7)
self.assertEqual(cooloff_time(None), timedelta(minutes=30))
def test_defaults_and_invalid_values(self):
for value in ("", "abc", "0", "-3"):
with self.subTest(value=value):
AppSettings.objects.filter(key__in=["LOGIN_FAILURE_LIMIT", "LOGIN_LOCKOUT_MINUTES"]).update(value=value)
self.assertEqual(failure_limit(None, {}), 5)
self.assertEqual(cooloff_time(None), timedelta(minutes=15))
def test_client_ip(self):
self.assertEqual(client_ip(self.request("127.0.0.1", "203.0.113.5")), "203.0.113.5")
self.assertEqual(client_ip(self.request("::1", "203.0.113.5")), "203.0.113.5")
self.assertEqual(client_ip(self.request("127.0.0.1")), "127.0.0.1")
self.assertEqual(client_ip(self.request("198.51.100.2", "203.0.113.5")), "198.51.100.2")
def test_client_ip_behind_a_configured_proxy(self):
with self.settings(LOGIN_TRUSTED_PROXIES=["172.18.0.0/16", "10.0.0.5"]):
self.assertEqual(client_ip(self.request("172.18.3.4", "203.0.113.5")), "203.0.113.5")
self.assertEqual(client_ip(self.request("10.0.0.5", "203.0.113.6")), "203.0.113.6")
self.assertEqual(client_ip(self.request("10.0.0.6", "203.0.113.6")), "10.0.0.6")
class SettingsPageTests(TestCase):
def test_both_values_are_on_the_settings_page(self):
self.client.force_login(get_user_model().objects.create_superuser("lo-admin", "lo@example.com", "pw"))
page = self.client.get(reverse("appsettings")).content.decode()
self.assertIn('name="LOGIN_FAILURE_LIMIT"', page)
self.assertIn('name="LOGIN_LOCKOUT_MINUTES"', page)
self.client.post(reverse("appsettings"), {"LOGIN_LOCKOUT_MINUTES": "45"})
self.assertEqual(cooloff_time(None), timedelta(minutes=45))

View file

@ -4,7 +4,7 @@ from django.conf import settings
from django.contrib.auth import get_user_model
from django.contrib.auth.models import Permission
from django.shortcuts import reverse
from django.test import Client, TestCase
from django.test import Client, RequestFactory, TestCase
from instances.models import Instance
@ -46,7 +46,7 @@ class AccountsTestCase(TestCase):
super().tearDownClass()
def setUp(self):
self.client.login(username="admin", password="admin")
self.client.login(request=RequestFactory().get("/"), username="admin", password="admin")
permission = Permission.objects.get(codename="change_password")
self.test_user.user_permissions.add(permission)
self.rsa_key = "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQC6OOdbfv27QVnSC6sKxGaHb6YFc+3gxCkyVR3cTSXE/n5BEGf8aOgBpepULWa1RZfxYHY14PlKULDygdXSdrrR2kNSwoKz/Oo4d+3EE92L7ocl1+djZbptzgWgtw1OseLwbFik+iKlIdqPsH+IUQvX7yV545ZQtAP8Qj1R+uCqkw== test@test"
@ -107,7 +107,7 @@ class AccountsTestCase(TestCase):
self.client.logout()
logged_in = self.client.login(username="test", password="newpw")
logged_in = self.client.login(request=RequestFactory().get("/"), username="test", password="newpw")
self.assertTrue(logged_in)
def test_user_instance_create_update_delete(self):

View file

@ -1,14 +1,14 @@
from django.contrib.auth.models import Group, User
from django.core.exceptions import ObjectDoesNotExist
from django.shortcuts import reverse
from django.test import TestCase
from django.test import RequestFactory, TestCase
from accounts.models import UserAttributes
class AdminTestCase(TestCase):
def setUp(self):
self.client.login(username="admin", password="admin")
self.client.login(request=RequestFactory().get("/"), username="admin", password="admin")
def test_group_list(self):
response = self.client.get(reverse("admin:group_list"))

View file

@ -0,0 +1,30 @@
from django.db import migrations
SETTINGS = [
("Login Failure Limit", "LOGIN_FAILURE_LIMIT", "5",
"Failed logins of a user from one address before it is locked"),
("Login Lockout Minutes", "LOGIN_LOCKOUT_MINUTES", "15",
"How long a locked login stays locked, in minutes"),
]
def add_settings(apps, schema_editor):
setting = apps.get_model("appsettings", "AppSettings")
db_alias = schema_editor.connection.alias
for name, key, value, description in SETTINGS:
setting.objects.using(db_alias).get_or_create(
key=key, defaults={"name": name, "value": value, "choices": "", "description": description}
)
def del_settings(apps, schema_editor):
setting = apps.get_model("appsettings", "AppSettings")
db_alias = schema_editor.connection.alias
setting.objects.using(db_alias).filter(key__in=[key for _, key, _, _ in SETTINGS]).delete()
class Migration(migrations.Migration):
dependencies = [("appsettings", "0013_remove_qemu_console_default_type")]
operations = [migrations.RunPython(add_settings, del_settings)]

View file

@ -1,4 +1,4 @@
from django.test import TestCase
from django.test import RequestFactory, TestCase
from django.urls import reverse
from appsettings.models import AppSettings
@ -7,7 +7,7 @@ from appsettings.settings import app_settings, get_settings
class AppSettingsTestCase(TestCase):
def setUp(self):
self.client.login(username="admin", password="admin")
self.client.login(request=RequestFactory().get("/"), username="admin", password="admin")
def test_get_settings_populates_singleton(self):
AppSettings.objects.update_or_create(

View file

@ -1,7 +1,7 @@
import os
from django.core.exceptions import ObjectDoesNotExist
from django.shortcuts import reverse
from django.test import TestCase, TransactionTestCase
from django.test import RequestFactory, TestCase, TransactionTestCase
from vrtManager.connection import CONN_SOCKET, connection_manager
from .models import Compute
@ -14,7 +14,7 @@ TEST_COMPUTE_TYPE = int(os.environ.get("TEST_LIBVIRT_TYPE", CONN_SOCKET))
class ComputesTestCase(TestCase):
def setUp(self):
self.client.login(username="admin", password="admin")
self.client.login(request=RequestFactory().get("/"), username="admin", password="admin")
self.compute, _ = Compute.objects.get_or_create(
name="computes-test-compute",
defaults={

View file

@ -4,6 +4,7 @@ django-bootstrap-icons==0.9.0
django-otp==1.7.3
django-qr-code==4.2.0
django-auth-ldap==5.3.0
django-axes==8.3.1
djangorestframework==3.17.2
drf-nested-routers==0.95.3
drf-spectacular[sidecar]==0.30.0

View file

@ -14,7 +14,7 @@ from datasource.views import get_client_ip, get_hostname_by_ip
class DataSourceTestCase(TestCase):
def setUp(self):
self.factory = RequestFactory()
self.client.login(username="admin", password="admin")
self.client.login(request=RequestFactory().get("/"), username="admin", password="admin")
self.compute = Compute.objects.create(
name="test_compute",
hostname="localhost",

View file

@ -2,7 +2,7 @@ import json
import socket
from unittest.mock import MagicMock, patch
from django.test import TestCase
from django.test import RequestFactory, TestCase
from django.urls import reverse
from computes.models import Compute
@ -11,7 +11,7 @@ from instances.models import Instance
class SetRootPassTestCase(TestCase):
def setUp(self):
self.client.login(username="admin", password="admin")
self.client.login(request=RequestFactory().get("/"), username="admin", password="admin")
self.compute = Compute.objects.create(
name="test_compute_root_pass",
hostname="localhost",

View file

@ -9,7 +9,7 @@ from django.contrib.auth import get_user_model
from django.contrib.auth.models import Permission
from django.http.response import Http404
from django.shortcuts import reverse
from django.test import TestCase
from django.test import RequestFactory, TestCase
from vrtManager.create import wvmCreate
from vrtManager.util import randomUUID
@ -121,7 +121,7 @@ class InstancesTestCase(TestCase):
)
def setUp(self):
self.client.login(username="admin", password="admin")
self.client.login(request=RequestFactory().get("/"), username="admin", password="admin")
self.rsa_key = "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQC6OOdbfv27QVnSC6sKxGaHb6YFc+3gxCkyVR3cTSXE/n5BEGf8aOgBpepULWa1RZfxYHY14PlKULDygdXSdrrR2kNSwoKz/Oo4d+3EE92L7ocl1+djZbptzgWgtw1OseLwbFik+iKlIdqPsH+IUQvX7yV545ZQtAP8Qj1R+uCqkw== test@test"
def test_index(self):

View file

@ -1,6 +1,6 @@
import json
from django.contrib.auth import get_user_model
from django.test import Client, TestCase
from django.test import Client, RequestFactory, TestCase
from django.urls import reverse
from computes.models import Compute
@ -40,7 +40,7 @@ class LogsTestCase(TestCase):
def test_vm_logs_superuser_access(self):
addlogmsg("admin", "localhost", "test-vm", "Created VM")
client = Client()
client.login(username="admin_logs", password="adminpassword")
client.login(request=RequestFactory().get("/"), username="admin_logs", password="adminpassword")
response = client.get(reverse("vm_logs", args=["test-vm"]))
self.assertEqual(response.status_code, 200)
@ -54,7 +54,7 @@ class LogsTestCase(TestCase):
def test_vm_logs_non_superuser_forbidden(self):
client = Client()
client.login(username="normal_logs", password="userpassword")
client.login(request=RequestFactory().get("/"), username="normal_logs", password="userpassword")
response = client.get(reverse("vm_logs", args=["test-vm"]))
# superuser_only decorator redirects non-superusers to index or login

View file

@ -5,7 +5,7 @@ import time
from unittest.mock import MagicMock, patch
from django.core.files.uploadedfile import SimpleUploadedFile
from django.test import TestCase
from django.test import RequestFactory, TestCase
from django.urls import reverse
from django.contrib.auth import get_user_model
@ -659,7 +659,7 @@ class StorageUploadSecurityTests(TestCase):
class StorageUploadViewTests(TestCase):
def setUp(self):
self.admin = User.objects.create_superuser(username="storage_admin", password="password", email="storage_admin@example.com")
self.client.login(username="storage_admin", password="password")
self.client.login(request=RequestFactory().get("/"), username="storage_admin", password="password")
self.compute = Compute.objects.create(name="LocalCompute", hostname="localhost", type=1, login="u", password="p")
@patch("storages.views.wvmStorage")
@ -830,7 +830,7 @@ class StorageUploadViewTests(TestCase):
# 4. Upload with same upload_id but by a different superuser cannot bypass/claim
User.objects.create_superuser(username="other_admin", password="password", email="other@example.com")
self.client.login(username="other_admin", password="password")
self.client.login(request=RequestFactory().get("/"), username="other_admin", password="password")
res4 = self.client.post(url, {
"iso_upload": "1",
"file": chunk_retry,

View file

@ -66,6 +66,7 @@ INSTALLED_APPS = [
"logs",
"qr_code",
"rest_framework",
"axes",
]
MIDDLEWARE = [
@ -84,6 +85,8 @@ MIDDLEWARE = [
"django.middleware.clickjacking.XFrameOptionsMiddleware",
"appsettings.middleware.AppSettingsMiddleware",
"webvirtcloud.middleware.ExceptionMiddleware",
# last, as django-axes requires
"axes.middleware.AxesMiddleware",
]
ROOT_URLCONF = "webvirtcloud.urls"
@ -124,12 +127,27 @@ DATABASES = {
}
AUTHENTICATION_BACKENDS = [
# first: refuses logins that are locked out
"axes.backends.AxesStandaloneBackend",
"django.contrib.auth.backends.ModelBackend",
#"django_auth_ldap.backend.LDAPBackend",
]
LOGIN_URL = "/accounts/login/"
# Failed logins (django-axes): a user is locked for the address the logins
# come from. The limit and the lock time are on the Settings page
# (LOGIN_FAILURE_LIMIT, LOGIN_LOCKOUT_MINUTES); unlock with
# "manage.py axes_reset_username <username>".
AXES_FAILURE_LIMIT = "accounts.lockout.failure_limit"
AXES_COOLOFF_TIME = "accounts.lockout.cooloff_time"
AXES_LOCKOUT_PARAMETERS = [["username", "ip_address"]]
AXES_RESET_ON_SUCCESS = True
AXES_CLIENT_IP_CALLABLE = "accounts.lockout.client_ip"
# Reverse proxies other than a local one whose X-Real-IP header gives the
# client address, e.g. ["10.0.0.5", "172.18.0.0/16"]
LOGIN_TRUSTED_PROXIES = []
LOGOUT_REDIRECT_URL = "/accounts/login/"
LOGIN_REDIRECT_URL="/instances/"