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

@@ -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.

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):
+def _migrated(state_apps, model, field=None):
-    """
+    """post_migrate follows every migrate, also targeted ones and rollbacks:
-    While a user still has the generated admin password, make them set a new
+    act only once the post-migration state has the model (and field)."""
-    one before using the panel (ROADMAP O-17; see ForcePasswordChangeMiddleware).
+    if state_apps is None:
-    """
+        return False
     try:
-        with open(admin_password_path()) as f:
+        fields = state_apps.get_model(model)._meta.get_fields()
-            generated = f.read().strip()
+    except LookupError:
-    except OSError:
+        return False
-        return
+    return field is None or any(f.name == field for f in fields)
-    if generated and user.check_password(generated):
-        request.session["must_change_password"] = True
 
 
 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
+    `docker logs` (ROADMAP O-02). Abort provisioning if private storage fails.
-    be written, so the admin is never locked out.
     """
     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:
+    except OSError as err:
-        print(f"\033[1m* \033[93mGenerated random admin password: {password}\033[0m")
+        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", [])
+    # Any real migrate run provisions while no user exists, so a run that
-    for migration, rolled_back in plan:
+    # failed to store the password can simply be repeated. flush (used by
-        if (
+    # tests) sends plan=None.
-            migration.app_label == "accounts"
+    if kwargs.get("plan") is None:
-            and migration.name == "0001_initial"
+        return
-            and not rolled_back
+    state_apps = kwargs.get("apps")
-        ):
+    if not (_migrated(state_apps, "auth.User") and _migrated(state_apps, "accounts.UserAttributes", "must_change_password")):
-            if User.objects.count() == 0:
+        return
-                is_testing = "test" in sys.argv
+    if User.objects.exists():
-                admin_user = os.environ.get("ADMIN_USERNAME", "admin")
+        return
-                admin_pass = os.environ.get("ADMIN_PASSWORD")
 
-                if not admin_pass:
+    is_testing = "test" in sys.argv
-                    if is_testing:
+    admin_user = os.environ.get("ADMIN_USERNAME", "admin")
-                        admin_pass = "admin"
+    admin_pass = os.environ.get("ADMIN_PASSWORD")
-                    else:
+    generated_password = not admin_pass and not is_testing
-                        admin_pass = secrets.token_urlsafe(16)
-                        _store_generated_password(admin_pass)
 
-                print(f"\033[1m* \033[92mCreating default admin user '{admin_user}'\033[0m")
+    if not admin_pass:
-                admin = User.objects.create_superuser(admin_user, None, admin_pass)
+        if is_testing:
-                UserAttributes(
+            admin_pass = "admin"
-                    user=admin,
+        else:
-                    max_instances=-1,
+            admin_pass = secrets.token_urlsafe(16)
-                    max_cpus=-1,
+            _store_generated_password(admin_pass)
-                    max_memory=-1,
+
-                    max_disk_size=-1,
+    print(f"\033[1m* \033[92mCreating default admin user '{admin_user}'\033[0m")
-                ).save()
+    with transaction.atomic():
-            break
+        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

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 []

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:

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),
+    ]

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,

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")
+        if plan is None:
-        migration.name = "0001_initial"
+            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())

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)

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()
+        with transaction.atomic():
-        update_session_auth_hash(request, user)  # Important!
+            user = form.save()
-        if request.session.pop("must_change_password", False):
+            UserAttributes.objects.filter(user=user).update(must_change_password=False)
-            # The generated first-install password is no longer valid.
+        update_session_auth_hash(request, user)
-            with contextlib.suppress(FileNotFoundError):
+        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")
 

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"]