From 6dabd69445e1424e01c5aa4e9b0fc36de6dd5d52 Mon Sep 17 00:00:00 2001 From: catborise Date: Sat, 3 Oct 2026 14:46:28 +0300 Subject: [PATCH] fix(accounts): first-login password change is stored, not in the session The forced change of the generated admin password was a session flag set at login when the password still matched data/admin_password. A new session after the file was gone bypassed it. - UserAttributes.must_change_password (migration 0007) holds it; the admin created with a generated password gets it, and a password change clears it. ForcePasswordChangeMiddleware reads it. - A system check (accounts.E001) stops the server unless the middleware follows AuthenticationMiddleware; the README has the settings change. - If the generated password cannot be stored privately, provisioning stops without printing it and without creating the admin (one transaction for the user and its attributes). - The admin is provisioned on any real migrate run while no user exists, so running migrate again after a failure creates it; flush (plan=None) and rollbacks of accounts are skipped. - The post_migrate receivers check the migration state for the models they use, so a targeted migrate on a fresh database (e.g. migrate appsettings) no longer fails with 'no such table'. --- README.md | 18 +++- accounts/apps.py | 97 ++++++++++--------- accounts/checks.py | 21 ++++ accounts/middleware.py | 6 +- ...007_userattributes_must_change_password.py | 44 +++++++++ accounts/models.py | 1 + accounts/test_admin_password.py | 46 ++++++++- accounts/test_first_login_password.py | 59 +++++++++++ accounts/views.py | 29 ++++-- admin/forms.py | 2 +- 10 files changed, 263 insertions(+), 60 deletions(-) create mode 100644 accounts/checks.py create mode 100644 accounts/migrations/0007_userattributes_must_change_password.py diff --git a/README.md b/README.md index 7c739472..b8a98174 100644 --- a/README.md +++ b/README.md @@ -394,6 +394,19 @@ environment=CSRF_TRUSTED_ORIGINS="https://webvirtcloud.example.com" ## How To Update +Before running migrations, add `accounts.middleware.ForcePasswordChangeMiddleware` +to `MIDDLEWARE` in your existing `webvirtcloud/settings.py`, after +`django.contrib.auth.middleware.AuthenticationMiddleware` (and the OTP/login-required +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. + +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 +requirement. If provisioning cannot write a new private password file, it stops +without creating an admin account or printing its password; fix the directory +permissions and rerun migrations. + ```bash # Go to Installation Directory cd /srv/webvirtcloud @@ -412,7 +425,10 @@ sudo systemctl restart supervisor # supervisord on RHEL / openSUSE ## Running Tests -WebVirtCloud includes unit tests for both Django models/views and the `vrtManager` libvirt abstraction layer. The test suite uses isolated mock drivers by default and does not require a live KVM hypervisor. +WebVirtCloud includes unit tests for both Django models/views and the `vrtManager` libvirt abstraction layer. The suite includes mocked regression tests and disk/clone tests using libvirt's +in-process `test:///default` driver. The live compute and instance integration +tests still require a configured libvirt host and are skipped when it is unavailable; +a green result with skips does not establish live KVM compatibility. ### 1. Setup Virtual Environment Use the same virtualenv as in [Local Development Setup](#local-development-setup), steps 1–3. Tests need the secret key too. diff --git a/accounts/apps.py b/accounts/apps.py index cdaedf70..2358bd20 100644 --- a/accounts/apps.py +++ b/accounts/apps.py @@ -1,8 +1,8 @@ import os from django.apps import AppConfig -from django.contrib.auth.signals import user_logged_in from django.db.models.signals import post_migrate +from django.db import transaction def admin_password_path(): @@ -11,18 +11,16 @@ def admin_password_path(): return os.path.join(str(settings.BASE_DIR), "data", "admin_password") -def flag_generated_password(sender, request, user, **kwargs): - """ - While a user still has the generated admin password, make them set a new - one before using the panel (ROADMAP O-17; see ForcePasswordChangeMiddleware). - """ +def _migrated(state_apps, model, field=None): + """post_migrate follows every migrate, also targeted ones and rollbacks: + act only once the post-migration state has the model (and field).""" + if state_apps is None: + return False try: - with open(admin_password_path()) as f: - generated = f.read().strip() - except OSError: - return - if generated and user.check_password(generated): - request.session["must_change_password"] = True + fields = state_apps.get_model(model)._meta.get_fields() + except LookupError: + return False + return field is None or any(f.name == field for f in fields) def apply_change_password(sender, **kwargs): @@ -33,6 +31,8 @@ def apply_change_password(sender, **kwargs): from django.conf import settings from django.contrib.auth.models import Permission, User + if not (_migrated(kwargs.get("apps"), "auth.Permission") and _migrated(kwargs.get("apps"), "auth.User")): + return if hasattr(settings, "SHOW_PROFILE_EDIT_PASSWORD"): print("\033[1m! \033[92mSHOW_PROFILE_EDIT_PASSWORD is found inside settings.py\033[0m") print("\033[1m* \033[92mApplying permission can_change_password for all users\033[0m") @@ -53,8 +53,7 @@ def _store_generated_password(password): """ Write the generated admin password to data/admin_password (mode 0600) instead of printing it, so it does not end up in install logs or - `docker logs` (ROADMAP O-02). Falls back to printing if the file cannot - be written, so the admin is never locked out. + `docker logs` (ROADMAP O-02). Abort provisioning if private storage fails. """ path = admin_password_path() try: @@ -64,12 +63,15 @@ def _store_generated_password(password): if os.path.lexists(path): os.unlink(path) fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) - os.fchmod(fd, 0o600) with os.fdopen(fd, "w") as f: + os.fchmod(f.fileno(), 0o600) f.write(password + "\n") print(f"\033[1m* \033[93mGenerated admin password written to {path}\033[0m") - except OSError: - print(f"\033[1m* \033[93mGenerated random admin password: {password}\033[0m") + except OSError as err: + raise RuntimeError( + f"Cannot securely store the generated admin password at {path}. " + "Fix the data directory permissions and run migrate again." + ) from err def create_admin(sender, **kwargs): @@ -83,35 +85,40 @@ def create_admin(sender, **kwargs): from accounts.models import UserAttributes - plan = kwargs.get("plan", []) - for migration, rolled_back in plan: - if ( - migration.app_label == "accounts" - and migration.name == "0001_initial" - and not rolled_back - ): - if User.objects.count() == 0: - is_testing = "test" in sys.argv - admin_user = os.environ.get("ADMIN_USERNAME", "admin") - admin_pass = os.environ.get("ADMIN_PASSWORD") + # Any real migrate run provisions while no user exists, so a run that + # failed to store the password can simply be repeated. flush (used by + # tests) sends plan=None. + if kwargs.get("plan") is None: + return + state_apps = kwargs.get("apps") + if not (_migrated(state_apps, "auth.User") and _migrated(state_apps, "accounts.UserAttributes", "must_change_password")): + return + if User.objects.exists(): + return - if not admin_pass: - if is_testing: - admin_pass = "admin" - else: - admin_pass = secrets.token_urlsafe(16) - _store_generated_password(admin_pass) + is_testing = "test" in sys.argv + admin_user = os.environ.get("ADMIN_USERNAME", "admin") + admin_pass = os.environ.get("ADMIN_PASSWORD") + generated_password = not admin_pass and not is_testing - print(f"\033[1m* \033[92mCreating default admin user '{admin_user}'\033[0m") - admin = User.objects.create_superuser(admin_user, None, admin_pass) - UserAttributes( - user=admin, - max_instances=-1, - max_cpus=-1, - max_memory=-1, - max_disk_size=-1, - ).save() - break + if not admin_pass: + if is_testing: + admin_pass = "admin" + else: + admin_pass = secrets.token_urlsafe(16) + _store_generated_password(admin_pass) + + print(f"\033[1m* \033[92mCreating default admin user '{admin_user}'\033[0m") + with transaction.atomic(): + admin = User.objects.create_superuser(admin_user, None, admin_pass) + UserAttributes.objects.create( + user=admin, + must_change_password=generated_password, + max_instances=-1, + max_cpus=-1, + max_memory=-1, + max_disk_size=-1, + ) class AccountsConfig(AppConfig): @@ -121,4 +128,4 @@ class AccountsConfig(AppConfig): def ready(self): post_migrate.connect(create_admin, sender=self) post_migrate.connect(apply_change_password, sender=self) - user_logged_in.connect(flag_generated_password) + from . import checks # noqa: F401 diff --git a/accounts/checks.py b/accounts/checks.py new file mode 100644 index 00000000..ca8ad8fe --- /dev/null +++ b/accounts/checks.py @@ -0,0 +1,21 @@ +from django.conf import settings +from django.core.checks import Error, Tags, register + + +@register(Tags.security) +def password_change_middleware_check(app_configs, **kwargs): + middleware = list(settings.MIDDLEWARE) + required = "accounts.middleware.ForcePasswordChangeMiddleware" + authentication = "django.contrib.auth.middleware.AuthenticationMiddleware" + if required not in middleware or ( + authentication in middleware + and middleware.index(required) < middleware.index(authentication) + ): + return [ + Error( + "ForcePasswordChangeMiddleware must follow AuthenticationMiddleware.", + hint=f"Add '{required}' to MIDDLEWARE in webvirtcloud/settings.py; see README.md.", + id="accounts.E001", + ) + ] + return [] diff --git a/accounts/middleware.py b/accounts/middleware.py index 445794d4..115d91b6 100644 --- a/accounts/middleware.py +++ b/accounts/middleware.py @@ -1,6 +1,8 @@ from django.shortcuts import redirect from django.urls import Resolver404, resolve +from accounts.models import UserAttributes + ALLOWED_VIEWS = {"accounts:change_password", "accounts:logout", "accounts:login"} @@ -11,7 +13,9 @@ class ForcePasswordChangeMiddleware: self.get_response = get_response def __call__(self, request): - if request.session.get("must_change_password"): + if request.user.is_authenticated and UserAttributes.objects.filter( + user=request.user, must_change_password=True + ).exists(): try: view_name = resolve(request.path_info).view_name except Resolver404: diff --git a/accounts/migrations/0007_userattributes_must_change_password.py b/accounts/migrations/0007_userattributes_must_change_password.py new file mode 100644 index 00000000..f4fd1f6d --- /dev/null +++ b/accounts/migrations/0007_userattributes_must_change_password.py @@ -0,0 +1,44 @@ +from pathlib import Path + +from django.conf import settings +from django.contrib.auth.hashers import check_password +from django.db import migrations, models + + +def flag_legacy_generated_password(apps, schema_editor): + """Preserve first-login enforcement for installations with a password file.""" + path = Path(settings.BASE_DIR) / "data" / "admin_password" + try: + generated = path.read_text().strip() + except FileNotFoundError: + return + except OSError as err: + raise RuntimeError( + f"Cannot read {path}; run migrate as its owner to preserve first-login enforcement." + ) from err + if not generated: + return + alias = schema_editor.connection.alias + User = apps.get_model(*settings.AUTH_USER_MODEL.split(".")) + Attributes = apps.get_model("accounts", "UserAttributes") + for user in User.objects.using(alias).filter(is_superuser=True): + if check_password(generated, user.password): + attributes, _ = Attributes.objects.using(alias).get_or_create( + user_id=user.pk + ) + attributes.must_change_password = True + attributes.save(using=alias, update_fields=["must_change_password"]) + + +class Migration(migrations.Migration): + dependencies = [ + ("accounts", "0006_alter_userattributes_id_alter_userinstance_id_and_more") + ] + operations = [ + migrations.AddField( + model_name="userattributes", + name="must_change_password", + field=models.BooleanField(default=False), + ), + migrations.RunPython(flag_legacy_generated_password, migrations.RunPython.noop), + ] diff --git a/accounts/models.py b/accounts/models.py index 189de683..a90b7b56 100644 --- a/accounts/models.py +++ b/accounts/models.py @@ -41,6 +41,7 @@ class UserSSHKey(models.Model): class UserAttributes(models.Model): user = models.OneToOneField(User, on_delete=models.CASCADE) can_clone_instances = models.BooleanField(default=True) + must_change_password = models.BooleanField(default=False) max_instances = models.IntegerField( _("max instances"), default=2, diff --git a/accounts/test_admin_password.py b/accounts/test_admin_password.py index eb9e7acc..a6c21639 100644 --- a/accounts/test_admin_password.py +++ b/accounts/test_admin_password.py @@ -7,23 +7,27 @@ from contextlib import redirect_stdout from pathlib import Path 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 accounts.apps import create_admin +from accounts.models import UserAttributes class GeneratedAdminPasswordTestCase(TestCase): - def _run_create_admin(self, base_dir): + def _run_create_admin(self, base_dir, plan=None): get_user_model().objects.all().delete() - migration = MagicMock(app_label="accounts") - migration.name = "0001_initial" + if plan is None: + migration = MagicMock(app_label="accounts") + migration.name = "0001_initial" + plan = [(migration, False)] out = io.StringIO() env = {k: v for k, v in os.environ.items() if k not in ("ADMIN_PASSWORD", "ADMIN_USERNAME")} with override_settings(BASE_DIR=Path(base_dir)), patch("sys.argv", ["manage.py", "migrate"]), patch.dict( os.environ, env, clear=True ), redirect_stdout(out): - create_admin(sender=None, plan=[(migration, False)]) + create_admin(sender=None, plan=plan, apps=apps) return out.getvalue() def test_password_is_written_to_a_private_file_not_printed(self): @@ -38,6 +42,7 @@ class GeneratedAdminPasswordTestCase(TestCase): self.assertNotIn(password, output) self.assertIn(str(password_file), output) self.assertIsNotNone(authenticate(username="admin", password=password)) + self.assertTrue(UserAttributes.objects.get(user__username="admin").must_change_password) def test_does_not_follow_a_planted_symlink(self): with tempfile.TemporaryDirectory() as tmp: @@ -52,3 +57,36 @@ class GeneratedAdminPasswordTestCase(TestCase): self.assertEqual(victim.read_text(), "original\n") self.assertFalse(password_file.is_symlink()) self.assertEqual(stat.S_IMODE(password_file.stat().st_mode), 0o600) + + def test_storage_failure_does_not_print_password_or_create_an_admin(self): + password = "review-dummy-generated-password" + output = io.StringIO() + with tempfile.TemporaryDirectory() as tmp, patch( + "secrets.token_urlsafe", return_value=password + ), patch("accounts.apps.os.open", side_effect=PermissionError("read-only")), redirect_stdout(output): + with self.assertRaisesRegex(RuntimeError, "Cannot securely store"): + self._run_create_admin(tmp) + self.assertNotIn(password, output.getvalue()) + self.assertFalse(get_user_model().objects.exists()) + + def test_rerunning_migrate_after_a_storage_failure_creates_the_admin(self): + with tempfile.TemporaryDirectory() as tmp: + with patch("accounts.apps.os.open", side_effect=PermissionError("read-only")), redirect_stdout(io.StringIO()): + with self.assertRaises(RuntimeError): + self._run_create_admin(tmp) + # The second migrate has nothing left to apply: an empty plan. + (Path(tmp) / "data").mkdir(exist_ok=True) + self._run_create_admin(tmp, plan=[]) + self.assertTrue(get_user_model().objects.filter(username="admin").exists()) + + def test_flush_does_not_create_an_admin(self): + get_user_model().objects.all().delete() + create_admin(sender=None, plan=None, apps=apps) + self.assertFalse(get_user_model().objects.exists()) + + def test_migrating_another_app_before_auth_creates_nothing(self): + get_user_model().objects.all().delete() + state_without_auth = MagicMock() + state_without_auth.get_model.side_effect = LookupError + create_admin(sender=None, plan=[], apps=state_without_auth) + self.assertFalse(get_user_model().objects.exists()) diff --git a/accounts/test_first_login_password.py b/accounts/test_first_login_password.py index a6cc1e03..132e5ced 100644 --- a/accounts/test_first_login_password.py +++ b/accounts/test_first_login_password.py @@ -5,6 +5,9 @@ from pathlib import Path from django.contrib.auth import get_user_model from django.test import TestCase, override_settings from django.urls import reverse +from unittest.mock import patch + +from accounts.models import UserAttributes GENERATED = "Generated-Pass-123" @@ -23,6 +26,7 @@ class FirstLoginPasswordChangeTestCase(TestCase): self.admin = get_user_model().objects.create_superuser( username="firstadmin", email="firstadmin@example.com", password=GENERATED ) + 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) @@ -45,6 +49,7 @@ class FirstLoginPasswordChangeTestCase(TestCase): }, ) self.assertFalse(self.password_file.exists()) + self.assertFalse(UserAttributes.objects.get(user=self.admin).must_change_password) response = self.client.get(reverse("instances:index")) self.assertEqual(response.status_code, 200) @@ -55,3 +60,57 @@ class FirstLoginPasswordChangeTestCase(TestCase): self.client.login(username="otheradmin", password="Other-pass-789") self.assertEqual(self.client.get(reverse("instances:index")).status_code, 200) self.assertTrue(self.password_file.exists()) + + def test_missing_file_and_a_new_session_do_not_bypass_enforcement(self): + self.password_file.unlink() + self.client.force_login(self.admin) + self.assertRedirects( + self.client.get(reverse("accounts:profile")), + reverse("accounts:change_password"), fetch_redirect_response=False, + ) + + 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.assertRedirects( + self.client.get(reverse("accounts:profile")), + reverse("accounts:change_password"), fetch_redirect_response=False, + ) + + def test_cleanup_failure_does_not_undo_a_successful_password_change(self): + self.client.force_login(self.admin) + with patch("accounts.views.os.remove", side_effect=PermissionError("read-only")): + response = self.client.post(reverse("accounts:change_password"), { + "old_password": GENERATED, "new_password1": "A-new-strong-pass-456", + "new_password2": "A-new-strong-pass-456", + }) + self.assertEqual(response.status_code, 302) + self.assertFalse(UserAttributes.objects.get(user=self.admin).must_change_password) + self.admin.refresh_from_db() + self.assertTrue(self.admin.check_password("A-new-strong-pass-456")) + self.assertEqual(self.client.get(reverse("accounts:profile")).status_code, 200) + + def test_missing_middleware_is_reported_by_the_upgrade_check(self): + from accounts.checks import password_change_middleware_check + with override_settings(MIDDLEWARE=["django.contrib.auth.middleware.AuthenticationMiddleware"]): + errors = password_change_middleware_check(None) + self.assertEqual([error.id for error in errors], ["accounts.E001"]) + + def test_legacy_migration_flags_only_the_matching_account(self): + import importlib + from types import SimpleNamespace + from django.apps import apps + from django.db import connection + migration = importlib.import_module("accounts.migrations.0007_userattributes_must_change_password") + UserAttributes.objects.filter(user=self.admin).update(must_change_password=False) + other = get_user_model().objects.create_superuser("legacy_other", "", "other-password") + migration.flag_legacy_generated_password(apps, SimpleNamespace(connection=connection)) + self.assertTrue(UserAttributes.objects.get(user=self.admin).must_change_password) + self.assertFalse(UserAttributes.objects.filter(user=other, must_change_password=True).exists()) + + def test_legacy_migration_fails_if_password_file_cannot_be_read(self): + import importlib + migration = importlib.import_module("accounts.migrations.0007_userattributes_must_change_password") + with patch("pathlib.Path.read_text", side_effect=PermissionError("unreadable")): + with self.assertRaisesRegex(RuntimeError, "run migrate as its owner"): + migration.flag_legacy_generated_password(None, None) diff --git a/accounts/views.py b/accounts/views.py index 6154892b..eb4d2af3 100644 --- a/accounts/views.py +++ b/accounts/views.py @@ -1,12 +1,14 @@ -import contextlib +import logging import os from admin.decorators import superuser_only from django.conf import settings from django.contrib import messages from django.contrib.auth import get_user_model, update_session_auth_hash, login as auth_login -from django.contrib.auth.decorators import permission_required +from django.contrib.auth.decorators import login_required from django.contrib.auth.forms import PasswordChangeForm +from django.core.exceptions import PermissionDenied +from django.db import transaction from django.shortcuts import get_object_or_404, redirect, render from django.urls import reverse from django.http import HttpResponseRedirect @@ -112,17 +114,28 @@ def account(request, user_id): ) -@permission_required("accounts.change_password", raise_exception=True) +@login_required def change_password(request): + must_change = UserAttributes.objects.filter( + user=request.user, must_change_password=True + ).exists() + if not must_change and not request.user.has_perm("accounts.change_password"): + raise PermissionDenied form = PasswordChangeForm(request.user, request.POST or None) if form.is_valid(): - user = form.save() - update_session_auth_hash(request, user) # Important! - if request.session.pop("must_change_password", False): - # The generated first-install password is no longer valid. - with contextlib.suppress(FileNotFoundError): + with transaction.atomic(): + user = form.save() + UserAttributes.objects.filter(user=user).update(must_change_password=False) + update_session_auth_hash(request, user) + request.session.pop("must_change_password", None) + if must_change and not UserAttributes.objects.filter(must_change_password=True).exists(): + try: os.remove(admin_password_path()) + except FileNotFoundError: + pass + except OSError: + logging.getLogger(__name__).warning("Could not remove the generated admin password file") messages.success(request, _("Password Changed")) return redirect("accounts:profile") diff --git a/admin/forms.py b/admin/forms.py index 140e6a8e..3f79ddce 100644 --- a/admin/forms.py +++ b/admin/forms.py @@ -114,4 +114,4 @@ class UserCreateForm(UserForm): class UserAttributesForm(forms.ModelForm): class Meta: model = UserAttributes - exclude = ["user", "can_clone_instances"] + exclude = ["user", "can_clone_instances", "must_change_password"]