Edward/password expiration - OP-3113

README.md

@@ -147,8 +147,10 @@ def schedule_tasks(scheduler: BackgroundScheduler): # Has to accept BackgroundSc
         max_instances=1
     )    
 ```
-Task will be automatically registered, but it will not be triggered unless `SCHEDULER_AUTOSTART` setting is
-set to `True`.
+Task will be automatically registered, but it will not be triggered unless the scheduler is running.
+For production deployments, run scheduled tasks in one dedicated scheduler process, for example
+`python manage.py runapscheduler`. Do not enable in-process scheduler autostart on gunicorn/web
+replicas, because each replica/worker could register and run the same jobs.
 
 ### Graphene Custom Types & Helper Classes/Methods
 * schema.SmallInt: Integer, with values ranging from -32768 to +32767

core/apps.py

@@ -29,6 +29,7 @@
         else "False"
     ),
     "password_reset_template": "password_reset.txt",
+    "password_expiry_reminder_template": "password_expiry_reminder.txt",
     "currency": "$",
     "gql_query_claim_admins_perms": [],
     "gql_query_users_perms": ["121701"],
@@ -58,6 +59,13 @@
     "is_valid_health_facility_contract_required": False,
     "secondary_calendar": None,
     "locked_user_password_hash": "locked",
+    "password_validity_days": 90,
+    "password_reuse_limit": 5,
+    "password_expiry_warning_days": 5,
+    "password_expiry_email_reminder_days": 5,
+    "password_expiry_email_reminder_hour": 8,
+    "password_expiry_email_reminder_minute": 0,
+    "password_expiry_email_reminder_timezone": "Africa/Blantyre",
     "gql_query_enable_viewing_masked_data_perms": ["900101"],
     "csrf_protect_login": True,
 }
@@ -70,6 +78,7 @@ class CoreConfig(AppConfig):
     username_changeable = True
     age_of_majority = 18
     password_reset_template = "password_reset.txt"
+    password_expiry_reminder_template = "password_expiry_reminder.txt"
 
     gql_query_claim_admins_perms = []
     gql_query_roles_perms = []
@@ -95,6 +104,13 @@ class CoreConfig(AppConfig):
     gql_mutation_delete_claim_administrator_perms = []
     is_valid_health_facility_contract_required = None
     locked_user_password_hash = None
+    password_validity_days = 90
+    password_reuse_limit = 5
+    password_expiry_warning_days = 5
+    password_expiry_email_reminder_days = 5
+    password_expiry_email_reminder_hour = 8
+    password_expiry_email_reminder_minute = 0
+    password_expiry_email_reminder_timezone = "Africa/Blantyre"
 
     fields_controls_user = {}
     fields_controls_eo = {}
@@ -260,7 +276,25 @@ def ready(self):
         self._configure_additional_settings(cfg)
 
         CoreConfig.password_reset_template = cfg["password_reset_template"]
+        CoreConfig.password_expiry_reminder_template = cfg[
+            "password_expiry_reminder_template"
+        ]
         CoreConfig.locked_user_password_hash = cfg["locked_user_password_hash"]
+        CoreConfig.password_validity_days = int(cfg["password_validity_days"])
+        CoreConfig.password_reuse_limit = int(cfg["password_reuse_limit"])
+        CoreConfig.password_expiry_warning_days = int(cfg["password_expiry_warning_days"])
+        CoreConfig.password_expiry_email_reminder_days = int(
+            cfg["password_expiry_email_reminder_days"]
+        )
+        CoreConfig.password_expiry_email_reminder_hour = int(
+            cfg["password_expiry_email_reminder_hour"]
+        )
+        CoreConfig.password_expiry_email_reminder_minute = int(
+            cfg["password_expiry_email_reminder_minute"]
+        )
+        CoreConfig.password_expiry_email_reminder_timezone = cfg[
+            "password_expiry_email_reminder_timezone"
+        ]
 
         # The scheduler starts as soon as it gets a job, which could be before Django is ready, so we enable it here
         from core import scheduler

core/migrations/0036_passwordexpiryreminderlog.py

@@ -0,0 +1,46 @@
+# Generated by Codex on 2026-07-20
+
+from django.db import migrations, models
+import django.db.models.deletion
+import django.utils.timezone
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("core", "0035_migrate_admin_users_to_superuser"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="PasswordExpiryReminderLog",
+            fields=[
+                (
+                    "id",
+                    models.AutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                ("password_validity", models.DateTimeField()),
+                ("reminder_date", models.DateField()),
+                ("sent_at", models.DateTimeField(default=django.utils.timezone.now)),
+                ("email", models.EmailField(max_length=200)),
+                (
+                    "user",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="password_expiry_reminder_logs",
+                        to="core.user",
+                    ),
+                ),
+            ],
+            options={
+                "db_table": "core_PasswordExpiryReminderLog",
+                "managed": True,
+                "unique_together": {("user", "password_validity", "reminder_date")},
+            },
+        ),
+    ]

core/models/__init__.py

@@ -12,6 +12,7 @@
 _get_default_expire_date = user._get_default_expire_date
 User = user.User
 UserRole = user.UserRole
+PasswordExpiryReminderLog = user.PasswordExpiryReminderLog
 VersionedModel = versioned_model.VersionedModel
 BaseVersionedModel = versioned_model.BaseVersionedModel
 HistoryModel = history_model.HistoryModel
@@ -32,4 +33,4 @@
 ObjectMutation = base_mutation.ObjectMutation
 CachedManager = versioned_model.CachedManager
 OpenIMISModel = openimis_model.OpenIMISModel
-OpenIMISMigrationModel = openimis_model.OpenIMISMigrationModel
+OpenIMISMigrationModel = openimis_model.OpenIMISMigrationModel

core/models/user.py

@@ -12,9 +12,11 @@
     PermissionsMixin,
     Group,
 )
-from django.core.exceptions import ObjectDoesNotExist
+from django.core.exceptions import ObjectDoesNotExist, ValidationError
 from django.db import models
 from django.utils.crypto import salted_hmac
+from django.utils import timezone
+from django.utils.translation import gettext as _
 from graphql import ResolveInfo
 import core
 from hashlib import sha256
@@ -24,6 +26,7 @@
 from .base import ExtendableModel, Language, UUIDModel
 from .versioned_model import VersionedModel
 from .openimis_model import OpenIMISMigrationModel, OpenIMISHistoryMixin  # , OpenIMISModel
+from core.apps import CoreConfig
 from core.utils import to_list_permissions
 from rest_framework import exceptions
 
@@ -353,14 +356,54 @@ def is_imis_admin(self):
             cache.set("is_admin_" + str(self.id), is_admin, 600)
         return is_admin
 
+    def _matches_password_hash(self, raw_password, private_key, password_hash):
+        if not raw_password or not password_hash:
+            return False
+        pwd_hash = sha256()
+        pwd_hash.update(f"{raw_password.rstrip()}{private_key}".encode())
+        return pwd_hash.hexdigest().lower() == password_hash.lower()
+
+    def _password_was_used(self, raw_password):
+        if self._matches_password_hash(raw_password, self.private_key, self.password):
+            return True
+        if not self.pk:
+            return False
+
+        password_reuse_limit = max(0, int(CoreConfig.password_reuse_limit))
+        if password_reuse_limit:
+            password_history = self.history.exclude(password__isnull=True).order_by(
+                "-history_date"
+            )
+            password_history = password_history[:password_reuse_limit]
+        else:
+            password_history = self.history.none()
+
+        return any(
+            self._matches_password_hash(
+                raw_password,
+                history.private_key,
+                history.password,
+            )
+            for history in password_history
+        )
+
+    @property
+    def is_password_expired(self):
+        return bool(self.password_validity and self.password_validity <= timezone.now())
+
     def set_password(self, raw_password, private_key=token_hex(128)):
         validate_password(raw_password)
+        if self._password_was_used(raw_password):
+            raise ValidationError(_("core.password_already_used"))
         self.private_key = private_key
         pwd_hash = sha256()
         pwd_hash.update(f"{raw_password.rstrip()}{self.private_key}".encode())
         self.password = (
             pwd_hash.hexdigest().upper()
         )  # Legacy requires this to be uppercase
+        self.password_validity = timezone.now() + timedelta(
+            days=CoreConfig.password_validity_days
+        )
 
     def check_password(self, raw_password):
         from hashlib import sha256
@@ -878,6 +921,23 @@ class Meta:
         unique_together = (("user", "group"),)
 
 
+class PasswordExpiryReminderLog(models.Model):
+    user = models.ForeignKey(
+        User,
+        models.CASCADE,
+        related_name="password_expiry_reminder_logs",
+    )
+    password_validity = models.DateTimeField()
+    reminder_date = models.DateField()
+    sent_at = models.DateTimeField(default=timezone.now)
+    email = models.EmailField(max_length=200)
+
+    class Meta:
+        managed = True
+        db_table = "core_PasswordExpiryReminderLog"
+        unique_together = (("user", "password_validity", "reminder_date"),)
+
+
 def _get_default_expire_date():
     return py_datetime.now() + timedelta(days=1)
 

core/scheduled_tasks.py

@@ -0,0 +1,24 @@
+import logging
+from zoneinfo import ZoneInfo
+
+from apscheduler.triggers.cron import CronTrigger
+
+from core.apps import CoreConfig
+from core.services import send_password_expiry_reminders
+
+logger = logging.getLogger(__name__)
+
+
+def schedule_tasks(scheduler):
+    scheduler.add_job(
+        send_password_expiry_reminders,
+        trigger=CronTrigger(
+            hour=CoreConfig.password_expiry_email_reminder_hour,
+            minute=CoreConfig.password_expiry_email_reminder_minute,
+            timezone=ZoneInfo(CoreConfig.password_expiry_email_reminder_timezone),
+        ),
+        id="core_password_expiry_email_reminders",
+        max_instances=1,
+        replace_existing=True,
+    )
+    logger.info("Scheduled core password expiry email reminders")

core/scheduler.py

@@ -21,6 +21,10 @@ def schedule_tasks(task_scheduler):
     execution
     :param scheduler: scheduler to which we'll add the tasks
     """
+    from core.scheduled_tasks import schedule_tasks as schedule_core_tasks
+
+    schedule_core_tasks(task_scheduler)
+
     if settings.SCHEDULER_JOBS:
         for job in settings.SCHEDULER_JOBS:
             logger.debug("Scheduling job %s", job["method"])