models.py

import csv
import ipaddress
import logging
import os
import string
from datetime import timedelta
from io import StringIO

import django
import phonenumbers
import swapper
from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer
from django.conf import settings
from django.contrib.auth import get_user_model
from django.core.cache import cache
from django.core.exceptions import ObjectDoesNotExist, ValidationError
from django.core.mail import send_mail
from django.core.serializers.json import DjangoJSONEncoder
from django.db import models, transaction
from django.db.models import JSONField, ProtectedError, Q
from django.utils import timezone
from django.utils.crypto import get_random_string
from django.utils.timezone import now
from django.utils.translation import gettext_lazy as _
from model_utils.fields import AutoLastModifiedField
from openwisp_notifications.signals import notify
from phonenumber_field.modelfields import PhoneNumberField
from private_storage.fields import PrivateFileField

from openwisp_radius.registration import (
    REGISTRATION_METHOD_CHOICES,
    get_registration_choices,
)
from openwisp_radius.tasks import process_radius_batch
from openwisp_users.mixins import OrgMixin
from openwisp_utils.base import KeyField, TimeStampedEditableModel, UUIDModel
from openwisp_utils.fields import (
    FallbackBooleanChoiceField,
    FallbackCharChoiceField,
    FallbackCharField,
    FallbackPositiveIntegerField,
    FallbackTextField,
)

from .. import exceptions
from .. import settings as app_settings
from ..settings import (
    BATCH_DEFAULT_PASSWORD_LENGTH,
    BATCH_MAIL_MESSAGE,
    BATCH_MAIL_SENDER,
    BATCH_MAIL_SUBJECT,
    DEFAULT_PASSWORD_RESET_URL,
)
from ..utils import (
    SmsMessage,
    decode_byte_data,
    find_available_username,
    generate_sms_token,
    get_sms_default_valid_until,
    load_model,
    prefix_generate_users,
    validate_csvfile,
)
from .validators import ipv6_network_validator, password_reset_url_validator

logger = logging.getLogger(__name__)
User = get_user_model()

OPTIONAL_FIELD_CHOICES = (
    ("disabled", _("Disabled")),
    ("allowed", _("Allowed")),
    ("mandatory", _("Mandatory")),
)

RADOP_CHECK_TYPES = (
    ("=", "="),
    (":=", ":="),
    ("==", "=="),
    ("+=", "+="),
    ("!=", "!="),
    (">", ">"),
    (">=", ">="),
    ("<", "<"),
    ("<=", "<="),
    ("=~", "=~"),
    ("!~", "!~"),
    ("=*", "=*"),
    ("!*", "!*"),
)
RAD_NAS_TYPES = app_settings.EXTRA_NAS_TYPES + (
    ("Async", "Async"),
    ("Sync", "Sync"),
    ("ISDN Sync", "ISDN Sync"),
    ("ISDN Async V.120", "ISDN Async V.120"),
    ("ISDN Async V.110", "ISDN Async V.110"),
    ("Virtual", "Virtual"),
    ("PIAFS", "PIAFS"),
    ("HDLC Clear", "HDLC Clear"),
    ("Channel", "Channel"),
    ("X.25", "X.25"),
    ("X.75", "X.75"),
    ("G.3 Fax", "G.3 Fax"),
    ("SDSL", "SDSL - Symmetric DSL"),
    ("ADSL-CAP", "ADSL-CAP"),
    ("ADSL-DMT", "ADSL-DMT"),
    ("IDSL", "IDSL"),
    ("Ethernet", "Ethernet"),
    ("xDSL", "xDSL"),
    ("Cable", "Cable"),
    ("Wireless - Other", "Wireless - Other"),
    ("IEEE 802.11", "Wireless - IEEE 802.11"),
    ("Token-Ring", "Token-Ring"),
    ("FDDI", "FDDI"),
    ("Wireless - CDMA2000", "Wireless - CDMA2000"),
    ("Wireless - UMTS", "Wireless - UMTS"),
    ("Wireless - 1X-EV", "Wireless - 1X-EV"),
    ("IAPP", "IAPP"),
    ("FTTP", "FTTP"),
    ("IEEE 802.16", "Wireless - IEEE 802.16"),
    ("IEEE 802.20", "Wireless - IEEE 802.20"),
    ("IEEE 802.22", "Wireless - IEEE 802.22"),
    ("PPPoA", "PPPoA - PPP over ATM"),
    ("PPPoEoA", "PPPoEoA - PPP over Ethernet over ATM"),
    ("PPPoEoE", "PPPoEoE - PPP over Ethernet over Ethernet"),
    ("PPPoEoVLAN", "PPPoEoVLAN - PPP over Ethernet over VLAN"),
    ("PPPoEoQinQ", "PPPoEoQinQ - PPP over Ethernet over IEEE 802.1QinQ"),
    ("xPON", "xPON - Passive Optical Network"),
    ("Wireless - XGP", "Wireless - XGP"),
    ("WiMAX", " WiMAX Pre-Release 8 IWK Function"),
    ("WIMAX-WIFI-IWK", "WIMAX-WIFI-IWK: WiMAX WIFI Interworking"),
    ("WIMAX-SFF", "WIMAX-SFF: Signaling Forwarding Function for LTE/3GPP2"),
    ("WIMAX-HA-LMA", "WIMAX-HA-LMA: WiMAX HA and or LMA function"),
    ("WIMAX-DHCP", "WIMAX-DHCP: WIMAX DCHP service"),
    ("WIMAX-LBS", "WIMAX-LBS: WiMAX location based service"),
    ("WIMAX-WVS", "WIMAX-WVS: WiMAX voice service"),
    ("Other", "Other"),
)
RADOP_REPLY_TYPES = (("=", "="), (":=", ":="), ("+=", "+="))
_STRATEGIES = (("prefix", _("Generate from prefix")), ("csv", _("Import from CSV")))
_NOT_BLANK_MESSAGE = _("This field cannot be blank.")
_GET_IP_LIST_HELP_TEXT = _(
    "Comma separated list of IP addresses allowed to access freeradius API"
)
_GET_MOBILE_PREFIX_HELP_TEXT = _(
    "Comma separated list of international mobile prefixes "
    "allowed to register via the user registration API."
)
_GET_OPTIONAL_FIELDS_HELP_TEXT = _(
    "Whether this field should be disabled, allowed or mandatory "
    "in the user registration API."
)
_REGISTRATION_ENABLED_HELP_TEXT = _(
    "Whether the registration API endpoint should be enabled or not"
)
_SAML_REGISTRATION_ENABLED_HELP_TEXT = _(
    "Whether the registration using SAML should be enabled or not"
)
_MAC_ADDR_ROAMING_ENABLED_HELP_TEXT = _(
    "Whether the MAC address roaming should be enabled or not."
)
_SOCIAL_REGISTRATION_ENABLED_HELP_TEXT = _(
    "Whether the registration using social applications should be enabled or not"
)
_SMS_VERIFICATION_HELP_TEXT = _(
    "Whether users who sign up should be required to verify their mobile "
    "phone number via SMS"
)
_ORGANIZATION_HELP_TEXT = _("The user is not a member of this organization")
_IDENTITY_VERIFICATION_ENABLED_HELP_TEXT = _(
    "Whether identity verification is required at the time of user registration"
)
_COA_ENABLED_HELP_TEXT = _("Whether RADIUS Change Of Authoization (CoA) is enabled")
_LOGIN_URL_HELP_TEXT = _("Enter the URL where users can log in to the wifi service")
_STATUS_URL_HELP_TEXT = _("Enter the URL where users can log out from the wifi service")
_PASSWORD_RESET_URL_HELP_TEXT = _("Enter the URL where users can reset their password")
OPTIONAL_SETTINGS = app_settings.OPTIONAL_REGISTRATION_FIELDS


class AutoUsernameMixin(object):
    def clean(self):
        """
        automatically sets username
        """
        if (
            self.username
            and User.objects.filter(username=self.username).exists()
            and not self.user
        ):
            self.user = User.objects.get(username=self.username)

        if self.user:
            self.username = self.user.username
            if hasattr(self, "organization") and not self.user.is_member(
                self.organization
            ):
                raise ValidationError({"organization": _ORGANIZATION_HELP_TEXT})
        elif not self.username:
            raise ValidationError(
                {"username": _NOT_BLANK_MESSAGE, "user": _NOT_BLANK_MESSAGE}
            )
        return super().clean()


class AutoGroupnameMixin(object):
    def _set_groupname(self):
        if self.group:
            self.groupname = self.group.name

    def clean(self):
        """
        automatically sets groupname
        """
        if self.group and not self.group.pk:
            return
        super().clean()
        self._set_groupname()
        if not self.group and not self.groupname:
            raise ValidationError(
                {"groupname": _NOT_BLANK_MESSAGE, "group": _NOT_BLANK_MESSAGE}
            )

    def save(self, *args, **kwargs):
        self._set_groupname()
        return super().save(*args, **kwargs)


class AttributeValidationMixin(object):
    def _get_validation_queryset_kwargs(self):
        raise NotImplementedError

    def _get_error_message(self):
        raise NotImplementedError

    @property
    def _object_name(self):
        return (
            type(self).__name__.lower().replace("radius", "").replace("group", "group ")
        )

    def clean(self):
        """
        checks if the check or reply attribute is unique
        """
        model = type(self).__name__
        if (
            load_model(model)
            .objects.filter(**self._get_validation_queryset_kwargs())
            .exclude(pk=self.pk)
            .exists()
        ):
            raise ValidationError({"attribute": self._get_error_message()})
        return super().clean()


class UserAttributeValidationMixin(AttributeValidationMixin):
    def _get_validation_queryset_kwargs(self):
        kwargs = dict(user=self.user, attribute=self.attribute)
        org = getattr(self, "organization", None)
        # only add `organization` key if it exists
        if org:
            kwargs["organization"] = org
        return kwargs

    def _get_error_message(self):
        return _(
            "Another %(object_name)s for the same user and with "
            "the same attribute already exists."
        ) % {"object_name": self._object_name}


class GroupAttributeValidationMixin(AttributeValidationMixin):
    def _get_validation_queryset_kwargs(self):
        return dict(group=self.group, attribute=self.attribute)

    def _get_error_message(self):
        return _(
            "Another %(object_name)s for the same group and with "
            "the same attribute already exists."
        ) % {"object_name": self._object_name}


class AbstractRadiusCheck(
    OrgMixin, AutoUsernameMixin, UserAttributeValidationMixin, TimeStampedEditableModel
):
    username = models.CharField(
        verbose_name=_("username"),
        max_length=64,
        db_index=True,
        # blank values are forbidden with custom validation
        # because this field can left blank if the user
        # foreign key is filled (it will be auto-filled)
        blank=True,
    )
    value = models.CharField(verbose_name=_("value"), max_length=253)
    op = models.CharField(
        verbose_name=_("operator"),
        max_length=2,
        choices=RADOP_CHECK_TYPES,
        default=":=",
    )
    attribute = models.CharField(
        verbose_name=_("attribute"),
        max_length=64,
    )
    # the foreign key is not part of the standard freeradius schema
    user = models.ForeignKey(
        settings.AUTH_USER_MODEL, on_delete=models.CASCADE, blank=True, null=True
    )

    class Meta:
        db_table = "radcheck"
        verbose_name = _("check")
        verbose_name_plural = _("checks")
        abstract = True

    def __str__(self):
        return self.username


class AbstractRadiusReply(
    OrgMixin, AutoUsernameMixin, UserAttributeValidationMixin, TimeStampedEditableModel
):
    username = models.CharField(
        verbose_name=_("username"),
        max_length=64,
        db_index=True,
        # blank values are forbidden with custom validation
        # because this field can left blank if the user
        # foreign key is filled (it will be auto-filled)
        blank=True,
    )
    value = models.CharField(verbose_name=_("value"), max_length=253)
    op = models.CharField(
        verbose_name=_("operator"), max_length=2, choices=RADOP_REPLY_TYPES, default="="
    )
    attribute = models.CharField(verbose_name=_("attribute"), max_length=64)
    # the foreign key is not part of the standard freeradius schema
    user = models.ForeignKey(
        settings.AUTH_USER_MODEL, on_delete=models.CASCADE, blank=True, null=True
    )

    class Meta:
        db_table = "radreply"
        verbose_name = _("reply")
        verbose_name_plural = _("replies")
        abstract = True

    def __str__(self):
        return self.username


class AbstractRadiusAccounting(OrgMixin, models.Model):
    session_id = models.CharField(
        verbose_name=_("session ID"),
        max_length=64,
        db_column="acctsessionid",
        db_index=True,
    )
    unique_id = models.CharField(
        verbose_name=_("accounting unique ID"),
        max_length=32,
        db_column="acctuniqueid",
        unique=True,
        primary_key=True,
    )
    username = models.CharField(
        verbose_name=_("username"), max_length=64, db_index=True, null=True, blank=True
    )
    groupname = models.CharField(
        verbose_name=_("group name"), max_length=64, null=True, blank=True
    )
    realm = models.CharField(
        verbose_name=_("realm"), max_length=64, null=True, blank=True
    )
    nas_ip_address = models.GenericIPAddressField(
        verbose_name=_("NAS IP address"), db_column="nasipaddress", db_index=True
    )
    nas_port_id = models.CharField(
        verbose_name=_("NAS port ID"),
        max_length=15,
        db_column="nasportid",
        null=True,
        blank=True,
    )
    nas_port_type = models.CharField(
        verbose_name=_("NAS port type"),
        max_length=32,
        db_column="nasporttype",
        null=True,
        blank=True,
    )
    start_time = models.DateTimeField(
        verbose_name=_("start time"),
        db_column="acctstarttime",
        db_index=True,
        null=True,
        blank=True,
    )
    update_time = models.DateTimeField(
        verbose_name=_("update time"), db_column="acctupdatetime", null=True, blank=True
    )
    stop_time = models.DateTimeField(
        verbose_name=_("stop time"),
        db_column="acctstoptime",
        db_index=True,
        null=True,
        blank=True,
    )
    interval = models.IntegerField(
        verbose_name=_("interval"), db_column="acctinterval", null=True, blank=True
    )
    session_time = models.PositiveIntegerField(
        verbose_name=_("session time"),
        db_column="acctsessiontime",
        null=True,
        blank=True,
    )
    authentication = models.CharField(
        verbose_name=_("authentication"),
        max_length=32,
        db_column="acctauthentic",
        null=True,
        blank=True,
    )
    connection_info_start = models.CharField(
        verbose_name=_("connection info start"),
        max_length=50,
        db_column="connectinfo_start",
        null=True,
        blank=True,
    )
    connection_info_stop = models.CharField(
        verbose_name=_("connection info stop"),
        max_length=50,
        db_column="connectinfo_stop",
        null=True,
        blank=True,
    )
    input_octets = models.BigIntegerField(
        verbose_name=_("input octets"),
        db_column="acctinputoctets",
        null=True,
        blank=True,
    )
    output_octets = models.BigIntegerField(
        verbose_name=_("output octets"),
        db_column="acctoutputoctets",
        null=True,
        blank=True,
    )
    called_station_id = models.CharField(
        verbose_name=_("called station ID"),
        max_length=253,
        db_column="calledstationid",
        db_index=True,
        blank=True,
        null=True,
    )
    calling_station_id = models.CharField(
        verbose_name=_("calling station ID"),
        max_length=253,
        db_column="callingstationid",
        db_index=True,
        blank=True,
        null=True,
    )
    terminate_cause = models.CharField(
        verbose_name=_("termination cause"),
        max_length=32,
        db_column="acctterminatecause",
        blank=True,
        null=True,
    )
    service_type = models.CharField(
        verbose_name=_("service type"),
        max_length=32,
        db_column="servicetype",
        null=True,
        blank=True,
    )
    framed_protocol = models.CharField(
        verbose_name=_("framed protocol"),
        max_length=32,
        db_column="framedprotocol",
        null=True,
        blank=True,
    )
    framed_ip_address = models.GenericIPAddressField(
        verbose_name=_("framed IP address"),
        db_column="framedipaddress",
        # the default MySQL freeradius schema defines
        # this as NOT NULL but defaulting to empty string
        # but that wouldn't work on PostgreSQL
        null=True,
        blank=True,
    )
    framed_ipv6_address = models.GenericIPAddressField(
        verbose_name=_("framed IPv6 address"),
        db_column="framedipv6address",
        protocol="IPv6",
        null=True,
        blank=True,
    )
    framed_ipv6_prefix = models.CharField(
        verbose_name=_("framed IPv6 prefix"),
        max_length=44,
        db_column="framedipv6prefix",
        validators=[ipv6_network_validator],
        null=True,
        blank=True,
    )
    framed_interface_id = models.CharField(
        verbose_name=_("framed interface ID"),
        max_length=19,
        db_column="framedinterfaceid",
        null=True,
        blank=True,
    )
    delegated_ipv6_prefix = models.CharField(
        verbose_name=_("delegated IPv6 prefix"),
        max_length=44,
        db_column="delegatedipv6prefix",
        validators=[ipv6_network_validator],
        null=True,
        blank=True,
    )

    def save(self, *args, **kwargs):
        if not self.start_time:
            self.start_time = now()
        super(AbstractRadiusAccounting, self).save(*args, **kwargs)

    class Meta:
        db_table = "radacct"
        verbose_name = _("accounting")
        verbose_name_plural = _("accountings")
        abstract = True

    def __str__(self):
        return self.unique_id

    @classmethod
    def close_stale_sessions(cls, days=None, hours=None):
        if hours:
            delta = timedelta(hours=hours)
        elif days:
            delta = timedelta(days=days)
        else:
            raise ValueError("Missing `days` or `hours`")
        # determine limit date time
        older_than = timezone.now() - delta
        # If the "update_time" is recent, then the session is not closed
        # even when the "start_time" is older than the specified time.
        # The "start_time" of a session is only checked when the
        # "update_time" is not set.
        sessions = cls.objects.filter(
            Q(stop_time__isnull=True)
            & (
                Q(update_time__lt=older_than)
                | (Q(update_time=None) & Q(start_time__lt=older_than))
            )
        )
        for session in sessions.iterator():
            # calculate seconds in between two dates
            session.session_time = (now() - session.start_time).total_seconds()
            session.stop_time = now()
            session.update_time = session.stop_time
            session.terminate_cause = "Session-Timeout"
            session.save()

    @classmethod
    def _close_stale_sessions_on_nas_boot(cls, called_station_id):
        """
        Called during RADIUS Accounting-On.
        """
        if not called_station_id:
            return 0
        stale_sessions = cls.objects.filter(
            called_station_id=called_station_id,
            stop_time__isnull=True,
        )
        closed_count = stale_sessions.update(
            stop_time=now(), terminate_cause="NAS-Reboot"
        )
        return closed_count


class AbstractNas(OrgMixin, TimeStampedEditableModel):
    name = models.CharField(
        verbose_name=_("name"),
        max_length=128,
        help_text=_("NAS Name (or IP address)"),
        db_index=True,
        db_column="nasname",
    )
    short_name = models.CharField(
        verbose_name=_("short name"), max_length=32, db_column="shortname"
    )
    type = models.CharField(
        verbose_name=_("type"), max_length=30, default="other", choices=RAD_NAS_TYPES
    )
    ports = models.PositiveIntegerField(verbose_name=_("ports"), blank=True, null=True)
    secret = models.CharField(
        verbose_name=_("secret"), max_length=60, help_text=_("Shared Secret")
    )
    server = models.CharField(
        verbose_name=_("server"), max_length=64, blank=True, null=True
    )
    community = models.CharField(
        verbose_name=_("community"), max_length=50, blank=True, null=True
    )
    description = models.CharField(
        verbose_name=_("description"), max_length=200, null=True, blank=True
    )

    class Meta:
        db_table = "nas"
        verbose_name = _("NAS")
        verbose_name_plural = _("NAS")
        abstract = True

    def __str__(self):
        return self.name


class AbstractRadiusGroup(OrgMixin, TimeStampedEditableModel):
    """
    This is not part of the standard freeradius schema.
    It's added to facilitate the management of groups.
    """

    name = models.CharField(
        verbose_name=_("group name"), max_length=255, unique=True, db_index=True
    )
    description = models.CharField(
        verbose_name=_("description"), max_length=64, blank=True, null=True
    )
    _DEFAULT_HELP_TEXT = (
        "The default group is automatically assigned to new users; "
        "changing the default group has only effect on new users "
        "(existing users will keep being members of their current group)"
    )
    default = models.BooleanField(
        verbose_name=_("is default?"), help_text=_(_DEFAULT_HELP_TEXT), default=False
    )

    class Meta:
        verbose_name = _("group")
        verbose_name_plural = _("groups")
        abstract = True

    def __str__(self):
        return self.name

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._initial_default = self.default

    def clean(self):
        self.check_default()
        if not hasattr(self, "organization"):
            return
        if not self.name.startswith(f"{self.organization.slug}-"):
            self.name = f"{self.organization.slug}-{self.name}"

    def save(self, *args, **kwargs):
        result = super().save(*args, **kwargs)
        if self.default:
            self.set_default()
        # sync all related records
        if not self._state.adding:
            self.radiusgroupcheck_set.update(groupname=self.name)
            self.radiusgroupreply_set.update(groupname=self.name)
            self.radiususergroup_set.update(groupname=self.name)
        return result

    _DEFAULT_VALIDATION_ERROR = _(
        "There must be at least one default group present in "
        "the system. To change the default group, simply set "
        "as default the group you want to make the new default."
    )
    _DEFAULT_PROTECTED_ERROR = _("The default group cannot be deleted")

    def delete(self, *args, **kwargs):
        if self.default:
            raise ProtectedError(self._DEFAULT_PROTECTED_ERROR, self)
        return super().delete(*args, **kwargs)

    def set_default(self):
        """
        ensures there's only 1 default group
        (logic overridable via custom models)
        """
        queryset = self.get_default_queryset()
        if queryset.exists():
            queryset.update(default=False)

    def check_default(self):
        """
        ensures the default group cannot be undefaulted
        (logic overridable via custom models)
        """
        if not self.default and self._initial_default:
            raise ValidationError({"default": self._DEFAULT_VALIDATION_ERROR})

    def get_default_queryset(self):
        """
        looks for default groups excluding the current one
        overridable by openwisp-radius and other 3rd party apps
        """
        return self.__class__.objects.exclude(pk=self.pk).filter(
            default=True, organization_id=self.organization.pk
        )


class AbstractRadiusUserGroup(
    AutoGroupnameMixin, AutoUsernameMixin, TimeStampedEditableModel
):
    username = models.CharField(
        verbose_name=_("username"),
        max_length=64,
        db_index=True,
        # blank values are forbidden with custom validation
        # because this field can left blank if the user
        # foreign key is filled (it will be auto-filled)
        blank=True,
    )
    groupname = models.CharField(
        verbose_name=_("group name"),
        max_length=64,
        # blank values are forbidden with custom validation
        # because this field can left blank if the group
        # foreign key is filled (it will be auto-filled)
        blank=True,
    )
    priority = models.IntegerField(verbose_name=_("priority"), default=1)
    # the foreign keys are not part of the standard freeradius schema,
    # these are added here to facilitate the synchronization of the
    # records which are related in different tables
    user = models.ForeignKey(
        settings.AUTH_USER_MODEL, on_delete=models.CASCADE, blank=True, null=True
    )
    group = models.ForeignKey(
        "RadiusGroup", on_delete=models.CASCADE, blank=True, null=True
    )

    class Meta:
        db_table = "radusergroup"
        verbose_name = _("user group")
        verbose_name_plural = _("user groups")
        unique_together = ("user", "group")
        abstract = True

    def __str__(self):
        return str(self.username)


class AbstractRadiusGroupCheck(
    AutoGroupnameMixin, GroupAttributeValidationMixin, TimeStampedEditableModel
):
    groupname = models.CharField(
        verbose_name=_("group name"),
        max_length=64,
        db_index=True,
        # blank values are forbidden with custom validation
        # because this field can left blank if the group
        # foreign key is filled (it will be auto-filled)
        blank=True,
    )
    attribute = models.CharField(verbose_name=_("attribute"), max_length=64)
    op = models.CharField(
        verbose_name=_("operator"),
        max_length=2,
        choices=RADOP_CHECK_TYPES,
        default=":=",
    )
    value = models.CharField(verbose_name=_("value"), max_length=253)
    # the foreign key is not part of the standard freeradius schema
    group = models.ForeignKey(
        "RadiusGroup", on_delete=models.CASCADE, blank=True, null=True
    )

    class Meta:
        db_table = "radgroupcheck"
        verbose_name = _("group check")
        verbose_name_plural = _("group checks")
        abstract = True

    def __str__(self):
        return str(self.groupname)


class AbstractRadiusGroupReply(
    AutoGroupnameMixin, GroupAttributeValidationMixin, TimeStampedEditableModel
):
    groupname = models.CharField(
        verbose_name=_("group name"),
        max_length=64,
        db_index=True,
        # blank values are forbidden with custom validation
        # because this field can left blank if the group
        # foreign key is filled (it will be auto-filled)
        blank=True,
    )
    attribute = models.CharField(verbose_name=_("attribute"), max_length=64)
    op = models.CharField(
        verbose_name=_("operator"), max_length=2, choices=RADOP_REPLY_TYPES, default="="
    )
    value = models.CharField(verbose_name=_("value"), max_length=253)
    # the foreign key is not part of the standard freeradius schema
    group = models.ForeignKey(
        "RadiusGroup", on_delete=models.CASCADE, blank=True, null=True
    )

    class Meta:
        db_table = "radgroupreply"
        verbose_name = _("group reply")
        verbose_name_plural = _("group replies")
        abstract = True

    def __str__(self):
        return str(self.groupname)


class AbstractRadiusPostAuth(OrgMixin, UUIDModel):
    username = models.CharField(verbose_name=_("username"), max_length=64)
    password = models.CharField(
        verbose_name=_("password"), max_length=64, db_column="pass", blank=True
    )
    reply = models.CharField(verbose_name=_("reply"), max_length=32)
    called_station_id = models.CharField(
        verbose_name=_("called station ID"),
        max_length=253,
        db_column="calledstationid",
        blank=True,
        null=True,
    )
    calling_station_id = models.CharField(
        verbose_name=_("calling station ID"),
        max_length=253,
        db_column="callingstationid",
        blank=True,
        null=True,
    )
    date = models.DateTimeField(
        verbose_name=_("date"), db_column="authdate", auto_now_add=True
    )

    class Meta:
        db_table = "radpostauth"
        verbose_name = _("post auth")
        verbose_name_plural = _("post auth log")
        abstract = True

    def __str__(self):
        return str(self.username)


def _get_csv_file_location(instance, filename):
    return os.path.join(
        str(instance.organization.slug),
        "batch",
        str(instance.organization.pk),
        "csv",
        filename,
    )


class AbstractRadiusBatch(OrgMixin, TimeStampedEditableModel):
    PENDING = "pending"
    PROCESSING = "processing"
    COMPLETED = "completed"
    FAILED = "failed"

    BATCH_STATUS_CHOICES = (
        (PENDING, _("Pending")),
        (PROCESSING, _("Processing")),
        (COMPLETED, _("Completed")),
        (FAILED, _("Failed")),
    )

    strategy = models.CharField(
        _("strategy"),
        max_length=16,
        choices=_STRATEGIES,
        db_index=True,
        help_text=_("Import users from a CSV or generate using a prefix"),
    )
    status = models.CharField(
        max_length=16,
        choices=BATCH_STATUS_CHOICES,
        default=PENDING,
        db_index=True,
    )
    name = models.CharField(
        verbose_name=_("name"),
        max_length=128,
        help_text=_("A unique batch name"),
        db_index=True,
        unique=False,
    )
    users = models.ManyToManyField(
        settings.AUTH_USER_MODEL,
        blank=True,
        related_name="radius_batch",
        help_text=_("List of users uploaded in this batch"),
    )
    csvfile = PrivateFileField(
        null=True,
        blank=True,
        verbose_name="CSV",
        storage=app_settings.PRIVATE_STORAGE_INSTANCE,
        upload_to=_get_csv_file_location,
        help_text=_("The csv file containing the user details to be uploaded"),
        max_file_size=app_settings.MAX_CSV_FILE_SIZE,
    )
    prefix = models.CharField(
        _("prefix"),
        null=True,
        blank=True,
        max_length=20,
        help_text=_("Usernames generated will be of the format [prefix][number]"),
    )
    # List of usernames and passwords used to create PDF
    user_credentials = JSONField(
        null=True,
        blank=True,
        verbose_name="PDF",
        encoder=DjangoJSONEncoder,
    )
    expiration_date = models.DateField(
        verbose_name=_("expiration date"),
        null=True,
        blank=True,
        help_text=_("If left blank users will never expire"),
    )

    class Meta:
        db_table = "radbatch"
        unique_together = ("name", "organization")
        verbose_name = _("batch user creation")
        verbose_name_plural = _("batch user creation operations")
        abstract = True

    def __str__(self):
        return self.name

    def clean(self):
        if self.strategy == "csv" and not self.csvfile:
            raise ValidationError(
                {"csvfile": _("This field cannot be blank.")}, code="invalid"
            )
        if self.strategy == "prefix" and not self.prefix:
            raise ValidationError(
                {"prefix": _("This field cannot be blank.")}, code="invalid"
            )
        if self.strategy == "prefix" and self.prefix:
            valid_chars = string.ascii_letters + string.digits + "@.+-_"
            for char in self.prefix:
                if char not in valid_chars:
                    raise ValidationError(
                        {
                            "prefix": _(
                                "This value may contain only letters, numbers,"
                                " and `@/./`+/-/_ characters."
                            )
                        },
                        code="invalid",
                    )
        if (
            self.strategy == "csv"
            and self.prefix
            or self.strategy == "prefix"
            and self.csvfile
        ):
            # this case would happen only when using the internal API
            raise ValidationError(
                _("Mixing fields of different strategies"), code="invalid"
            )
        if self.strategy == "csv":
            validate_csvfile(self.csvfile.file)
        super().clean()

    def add(self, reader, password_length=BATCH_DEFAULT_PASSWORD_LENGTH):
        users_list = []
        generated_passwords = []
        with transaction.atomic():
            for row in reader:
                if len(row) == 5:
                    user, password = self.get_or_create_user(
                        row, users_list, password_length
                    )
                    users_list.append(user)
                    if password:
                        generated_passwords.append(password)
            for user in users_list:
                self.save_user(user)
        for element in generated_passwords:
            username, password, user_email = element
            send_mail(
                BATCH_MAIL_SUBJECT,
                BATCH_MAIL_MESSAGE.format(username, password),
                BATCH_MAIL_SENDER,
                [user_email],
            )

    def csvfile_upload(
        self, csvfile=None, password_length=BATCH_DEFAULT_PASSWORD_LENGTH
    ):
        if not csvfile:
            csvfile = self.csvfile
        csv_data = csvfile.read()
        csv_data = decode_byte_data(csv_data)
        reader = csv.reader(StringIO(csv_data), delimiter=",")
        with transaction.atomic():
            self.full_clean()
            self.save()
            self.add(reader, password_length)

    def prefix_add(self, prefix, n, password_length=BATCH_DEFAULT_PASSWORD_LENGTH):
        users_list, user_credentials = prefix_generate_users(prefix, n, password_length)
        for user in users_list:
            user.full_clean()
            self.save_user(user)
        self.user_credentials = user_credentials
        self.full_clean()
        self.save()

    def get_or_create_user(self, row, users_list, password_length):
        User = get_user_model()
        username, password, email, first_name, last_name = row
        if email and User.objects.filter(email=email).exists():
            user = User.objects.get(email=email)
            return user, None
        generated_password = None
        username, password, email, first_name, last_name = row
        if not username and email:
            username = email.split("@")[0]
        username = find_available_username(username, users_list)
        user = User(
            username=username, email=email, first_name=first_name, last_name=last_name
        )
        cleartext_delimiter = "cleartext$"
        if not password:
            password = get_random_string(length=password_length)
            user.set_password(password)
            generated_password = [username, password, email]
        elif password and password.startswith(cleartext_delimiter):
            password = password[len(cleartext_delimiter) :]
            user.set_password(password)
        else:
            user.password = password
        user.full_clean()
        return user, generated_password

    def save_user(self, user):
        OrganizationUser = swapper.load_model("openwisp_users", "OrganizationUser")
        RegisteredUser = swapper.load_model("openwisp_radius", "RegisteredUser")
        user.save()
        registered_user = RegisteredUser(user=user, method="manual")
        if self.organization.radius_settings.needs_identity_verification:
            registered_user.is_verified = True
        registered_user.save()
        self.users.add(user)
        if OrganizationUser.objects.filter(
            user=user, organization=self.organization
        ).exists():
            return
        obj = OrganizationUser(
            user=user, organization=self.organization, is_admin=False
        )
        obj.full_clean()
        obj.save()

    def delete(self):
        self.users.all().delete()
        super().delete()
        self._remove_files()

    def expire(self):
        users = self.users.all()
        for u in users:
            u.is_active = False
            u.save()

    def _remove_files(self):
        if self.csvfile:
            self.csvfile.storage.delete(self.csvfile.name)

    def schedule_processing(self, number_of_users=0):
        items_to_process = 0
        if self.strategy == "prefix":
            items_to_process = number_of_users
        elif self.strategy == "csv" and self.csvfile:
            try:
                csv_data = self.csvfile.read()
                decoded_data = decode_byte_data(csv_data)
                items_to_process = sum(1 for row in csv.reader(StringIO(decoded_data)))
                self.csvfile.seek(0)
            except Exception as e:
                logger.error(f"Could not count rows in CSV for batch {self.pk}: {e}")
                items_to_process = app_settings.BATCH_ASYNC_THRESHOLD
        is_async = items_to_process >= app_settings.BATCH_ASYNC_THRESHOLD
        if is_async:
            process_radius_batch.delay(self.pk, number_of_users=number_of_users)
        else:
            self.process(number_of_users=number_of_users)
        return is_async

    def process(self, number_of_users=0, is_async=False):
        channel_layer = get_channel_layer()
        group_name = f"radius_batch_{self.pk}"
        try:
            self.status = self.PROCESSING
            self.save(update_fields=["status"])
            if self.strategy == "prefix":
                self.prefix_add(self.prefix, number_of_users)
            elif self.strategy == "csv":
                self.csvfile_upload()
            self.status = self.COMPLETED
            self.save(update_fields=["status"])
            if is_async:
                notify.send(
                    type="generic_message",
                    level="success",
                    message=_(
                        f'The batch creation operation for "{self.name}" '
                        "has completed successfully."
                    ),
                    sender=self.organization,
                    target=self,
                    description=_(f"Number of users processed: {self.users.count()}."),
                )
        except Exception as e:
            logger.error(
                "RadiusBatch %s failed during processing:", self.pk, exc_info=True
            )
            self.status = self.FAILED
            self.save(update_fields=["status"])
            notify.send(
                type="generic_message",
                level="error",
                message=_(f'The batch creation operation for "{self.name}" failed.'),
                sender=self.organization,
                target=self,
                description=_(
                    f"An error occurred while processing the batch.\n\n" f"Error: {e}"
                ),
            )
        finally:
            async_to_sync(channel_layer.group_send)(
                group_name, {"type": "batch_status_update", "status": self.status}
            )


class AbstractRadiusToken(OrgMixin, TimeStampedEditableModel, models.Model):
    # key field is a primary key so additional id field will be redundant
    id = None
    # tokens are not supposed to be modified, can be regenerated if necessary
    modified = None
    key = models.CharField(_("Key"), max_length=40, primary_key=True)
    user = models.OneToOneField(
        settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="radius_token"
    )
    can_auth = models.BooleanField(
        default=False,
        help_text=(
            "Enable the radius token to be used for freeradius authorization request"
        ),
    )

    class Meta:
        db_table = "radiustoken"
        verbose_name = _("radius token")
        verbose_name_plural = _("radius token")
        abstract = True

    def save(self, *args, **kwargs):
        if not self.key:
            self.key = self.generate_key()
        return super().save(*args, **kwargs)

    def delete_cache(self, *args, **kwargs):
        username = f"rt-{self.user.username}"
        if cache.get(username):
            cache.delete(username)

    def generate_key(self):
        return get_random_string(length=40)

    def __str__(self):
        # When the object is deleted, the key is set to None.
        # This raises error when deleting the object from
        # UserAdmin.
        return self.key or f"RadiusToken: {self.user.username}"


class AbstractOrganizationRadiusSettings(UUIDModel):
    organization = models.OneToOneField(
        swapper.get_model_name("openwisp_users", "Organization"),
        verbose_name=_("organization"),
        related_name="radius_settings",
        on_delete=models.CASCADE,
    )
    token = KeyField(max_length=32)
    sms_verification = FallbackBooleanChoiceField(
        help_text=_SMS_VERIFICATION_HELP_TEXT,
        fallback=app_settings.SMS_VERIFICATION_ENABLED,
        verbose_name=_("SMS verification"),
    )
    needs_identity_verification = FallbackBooleanChoiceField(
        help_text=_IDENTITY_VERIFICATION_ENABLED_HELP_TEXT,
        fallback=app_settings.NEEDS_IDENTITY_VERIFICATION,
    )
    sms_sender = models.CharField(
        _("Sender"),
        max_length=128,
        blank=True,
        null=True,
        help_text=_(
            "alpha numeric identifier used as sender for SMS sent by this organization"
        ),
    )
    sms_message = FallbackTextField(
        _("SMS Message"),
        max_length=160,
        help_text=_(
            "SMS message template used for sending verification code."
            ' Must contain "{code}" placeholder for OTP value.'
        ),
        fallback=app_settings.SMS_MESSAGE_TEMPLATE,
    )
    sms_cooldown = FallbackPositiveIntegerField(
        _("SMS Cooldown"),
        help_text=_(
            "Time period a user will have to wait before requesting"
            " another SMS token (in seconds)."
        ),
        fallback=app_settings.SMS_COOLDOWN,
    )
    sms_meta_data = JSONField(
        null=True,
        blank=True,
        help_text=_(
            "Additional configuration for SMS backend in JSON format"
            " (optional, leave blank if unsure)"
        ),
        verbose_name=_("SMS meta data"),
        encoder=DjangoJSONEncoder,
    )
    freeradius_allowed_hosts = FallbackTextField(
        help_text=_GET_IP_LIST_HELP_TEXT,
        fallback=",".join(app_settings.FREERADIUS_ALLOWED_HOSTS),
    )
    coa_enabled = FallbackBooleanChoiceField(
        help_text=_COA_ENABLED_HELP_TEXT,
        fallback=app_settings.COA_ENABLED,
        verbose_name=_("CoA Enabled"),
    )
    allowed_mobile_prefixes = FallbackTextField(
        help_text=_GET_MOBILE_PREFIX_HELP_TEXT,
        fallback=",".join(app_settings.ALLOWED_MOBILE_PREFIXES),
    )
    first_name = FallbackCharChoiceField(
        verbose_name=_("first name"),
        help_text=_GET_OPTIONAL_FIELDS_HELP_TEXT,
        max_length=12,
        choices=OPTIONAL_FIELD_CHOICES,
        fallback=OPTIONAL_SETTINGS.get("first_name", None),
    )
    last_name = FallbackCharChoiceField(
        verbose_name=_("last name"),
        help_text=_GET_OPTIONAL_FIELDS_HELP_TEXT,
        max_length=12,
        choices=OPTIONAL_FIELD_CHOICES,
        fallback=OPTIONAL_SETTINGS.get("last_name", None),
    )
    location = FallbackCharChoiceField(
        verbose_name=_("location"),
        help_text=_GET_OPTIONAL_FIELDS_HELP_TEXT,
        max_length=12,
        choices=OPTIONAL_FIELD_CHOICES,
        fallback=OPTIONAL_SETTINGS.get("location", None),
    )
    birth_date = FallbackCharChoiceField(
        verbose_name=_("birth date"),
        help_text=_GET_OPTIONAL_FIELDS_HELP_TEXT,
        max_length=12,
        choices=OPTIONAL_FIELD_CHOICES,
        fallback=OPTIONAL_SETTINGS.get("birth_date", None),
    )
    registration_enabled = FallbackBooleanChoiceField(
        help_text=_REGISTRATION_ENABLED_HELP_TEXT,
        fallback=app_settings.REGISTRATION_API_ENABLED,
    )
    saml_registration_enabled = FallbackBooleanChoiceField(
        help_text=_SAML_REGISTRATION_ENABLED_HELP_TEXT,
        verbose_name=_("SAML registration enabled"),
        fallback=app_settings.SAML_REGISTRATION_ENABLED,
    )
    mac_addr_roaming_enabled = FallbackBooleanChoiceField(
        help_text=_MAC_ADDR_ROAMING_ENABLED_HELP_TEXT,
        verbose_name=_("MAC address roaming enabled"),
        fallback=app_settings.MAC_ADDR_ROAMING_ENABLED,
    )
    social_registration_enabled = FallbackBooleanChoiceField(
        help_text=_SOCIAL_REGISTRATION_ENABLED_HELP_TEXT,
        fallback=app_settings.SOCIAL_REGISTRATION_ENABLED,
    )
    login_url = models.URLField(
        verbose_name=_("Login URL"),
        null=True,
        blank=True,
        help_text=_LOGIN_URL_HELP_TEXT,
    )
    status_url = models.URLField(
        verbose_name=_("Status URL"),
        null=True,
        blank=True,
        help_text=_STATUS_URL_HELP_TEXT,
    )
    password_reset_url = FallbackCharField(
        verbose_name=_("Password reset URL"),
        max_length=200,
        help_text=_PASSWORD_RESET_URL_HELP_TEXT,
        fallback=DEFAULT_PASSWORD_RESET_URL,
        validators=[password_reset_url_validator],
    )

    class Meta:
        verbose_name = _("Organization radius settings")
        verbose_name_plural = verbose_name
        abstract = True

    def __str__(self):
        return self.organization.name

    @property
    def freeradius_allowed_hosts_list(self):
        addresses = []
        if self.freeradius_allowed_hosts:
            addresses = self.freeradius_allowed_hosts.split(",")
        return addresses

    @property
    def allowed_mobile_prefixes_list(self):
        mobile_prefixes = []
        if self.allowed_mobile_prefixes:
            mobile_prefixes = self.allowed_mobile_prefixes.split(",")
        return mobile_prefixes

    def clean(self):
        if self.sms_verification and not self.sms_sender:
            raise ValidationError(
                {
                    "sms_sender": _(
                        "if SMS verification is enabled this field is required."
                    )
                }
            )
        self._clean_freeradius_allowed_hosts()
        self._clean_allowed_mobile_prefixes()
        self._clean_password_reset_url()
        self._clean_sms_message()

    def _clean_freeradius_allowed_hosts(self):
        allowed_hosts_set = set(self.freeradius_allowed_hosts_list)
        settings_allowed_hosts_set = set(app_settings.FREERADIUS_ALLOWED_HOSTS)
        if not allowed_hosts_set and not settings_allowed_hosts_set:
            raise ValidationError(
                {
                    "freeradius_allowed_hosts": _(
                        "Cannot be empty when the settings value for "
                        "`OPENWISP_RADIUS_FREERADIUS_ALLOWED_HOSTS` is not provided."
                    )
                }
            )
        elif not allowed_hosts_set:
            self.freeradius_allowed_hosts = None
        elif allowed_hosts_set:
            if allowed_hosts_set == settings_allowed_hosts_set:
                self.freeradius_allowed_hosts = None
            else:
                try:
                    for ip_address in allowed_hosts_set:
                        ipaddress.ip_network(ip_address)
                except ValueError:
                    raise ValidationError(
                        {
                            "freeradius_allowed_hosts": _(
                                "Invalid input. Please enter valid ip addresses "
                                "or subnets separated by comma. (no spaces)"
                            )
                        }
                    )

    def _clean_allowed_mobile_prefixes(self):
        valid_country_codes = phonenumbers.COUNTRY_CODE_TO_REGION_CODE.keys()
        allowed_mobile_prefixes_set = set(self.allowed_mobile_prefixes_list)
        settings_allowed_mobile_prefixes_set = set(app_settings.ALLOWED_MOBILE_PREFIXES)
        for code in self.allowed_mobile_prefixes_list:
            if not code or code[0] != "+" or int(code[1:]) not in valid_country_codes:
                raise ValidationError(
                    {
                        "allowed_mobile_prefixes": _(
                            "Invalid input. Please enter valid mobile "
                            "prefixes separated by comma. (no spaces)"
                        )
                    }
                )

        if (
            not allowed_mobile_prefixes_set
            or allowed_mobile_prefixes_set == settings_allowed_mobile_prefixes_set
        ):
            self.allowed_mobile_prefixes = None

    def _clean_password_reset_url(self):
        if self.password_reset_url and (
            "{uid}" not in self.password_reset_url
            or "{token}" not in self.password_reset_url
        ):
            raise ValidationError(
                {
                    "password_reset_url": _(
                        'The URL must contain the "{{token}}" and '
                        '"{{uid}}" placeholders, eg: {}.'.format(
                            DEFAULT_PASSWORD_RESET_URL
                        )
                    )
                }
            )

    def _clean_sms_message(self):
        if self.sms_message and ("{code}" not in self.sms_message):
            raise ValidationError(
                {
                    "sms_message": _(
                        'The SMS message must contain the "{{code}}" '
                        "placeholder, eg: {}.".format(app_settings.SMS_MESSAGE_TEMPLATE)
                    )
                }
            )

    def save_cache(self, *args, **kwargs):
        cache.set(self.organization.pk, self.token)
        cache.set(f"ip-{self.organization.pk}", self.freeradius_allowed_hosts_list)

    def delete_cache(self, *args, **kwargs):
        cache.delete(self.organization.pk)
        cache.delete(f"ip-{self.organization.pk}")


class AbstractPhoneToken(TimeStampedEditableModel):
    """
    Phone Verification Token (sent via SMS)
    """

    user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
    valid_until = models.DateTimeField(default=get_sms_default_valid_until)
    attempts = models.PositiveIntegerField(default=0)
    verified = models.BooleanField(default=False)
    token = models.CharField(max_length=8, editable=False, default=generate_sms_token)
    ip = models.GenericIPAddressField()
    # if users change their phone number, a record of the previously
    # verified phone numbers is kept thanks to this field
    phone_number = PhoneNumberField(blank=False, null=False)

    class Meta:
        verbose_name = _("Phone verification token")
        verbose_name_plural = _("Phone verification tokens")
        ordering = ("-created",)
        indexes = [
            models.Index(fields=["user", "created"]),
            models.Index(fields=["user", "created", "ip"]),
        ]
        abstract = True

    def clean(self):
        if not hasattr(self, "user"):
            return
        self._validate_phone_number_uniqueness()
        self._validate_max_attempts()

    def _validate_phone_number_uniqueness(self):
        """
        Users cannot create a phone token if the number
        is already taken by another user
        """
        phone_number_already_taken = (
            User.objects.filter(phone_number=self.phone_number)
            .exclude(pk=self.user.pk)
            .exists()
        )
        if phone_number_already_taken:
            raise ValidationError(
                {
                    "phone_number": _(
                        "This phone number is already associated to another user."
                    )
                }
            )

    def _validate_max_attempts(self):
        """
        Enforce limits on the creation of phone tokens to prevent abuse
        which can lead to excessive expenditure for sending SMS
        """
        date_start = timezone.localdate()
        date_end = date_start + timedelta(days=1)
        PhoneToken = load_model("PhoneToken")
        qs = PhoneToken.objects.filter(created__range=[date_start, date_end])
        # limit generation of tokens per day by user
        user_token_count = qs.filter(user=self.user).count()
        if user_token_count >= app_settings.SMS_TOKEN_MAX_USER_DAILY:
            logger.warning(
                f"The user {self.user} has reached the maximum daily SMS limit."
            )
            raise ValidationError(_("Maximum daily limit reached."))
        # limit generation of tokens per day by ip
        ip_token_count = qs.filter(ip=self.ip).count()
        if ip_token_count >= app_settings.SMS_TOKEN_MAX_IP_DAILY:
            logger.warning(
                logger.warning(
                    _(
                        f"User {self.user} has reached the maximum "
                        f"daily SMS limit from ip address {self.ip}"
                    )
                )
            )
            raise ValidationError(
                _("Maximum daily limit reached from this IP address.")
            )

    def save(self, *args, **kwargs):
        created = self._state.adding
        result = super().save(*args, **kwargs)
        if created:
            self.send_token()
        return result

    def send_token(self):
        OrganizationUser = swapper.load_model("openwisp_users", "OrganizationUser")
        org_user = OrganizationUser.objects.filter(user=self.user).first()
        if not org_user:
            raise exceptions.NoOrgException(
                _("The user {user} is not member of any organization").format(
                    user=self.user
                )
            )
        org_radius_settings = org_user.organization.radius_settings
        message = _(org_radius_settings.sms_message).format(
            organization=org_radius_settings.organization.name, code=self.token
        )
        sms_message = SmsMessage(
            body=str(message),
            from_phone=str(org_radius_settings.sms_sender),
            to=[str(self.phone_number)],
        )
        sms_message.send(meta_data=org_radius_settings.sms_meta_data)

    def is_valid(self, token):
        self.attempts += 1
        try:
            self.verified = self.__check(token)
        except exceptions.PhoneTokenException as phone_error:
            self.save()
            raise phone_error
        self.save()
        return self.verified

    def _validate_already_verified(self):
        try:
            if self.user.registered_user.is_verified:
                logger.warning(f"User {self.user.pk} is already verified")
                raise exceptions.UserAlreadyVerified(
                    _("This user has been already verified.")
                )
        except ObjectDoesNotExist:
            pass

    def __check(self, token):
        self._validate_already_verified()
        if self.attempts > app_settings.SMS_TOKEN_MAX_ATTEMPTS:
            logger.warning(
                f"User {self.user} has reached the max "
                f"attempt limit for token {self.pk}."
            )
            raise exceptions.MaxAttemptsException(
                _("Maximum number of attempts reached, request a new code.")
            )
        if timezone.now() > self.valid_until:
            logger.warning(
                f"User {self.user} has tried to verify an expired token: {self.pk}."
            )
            raise exceptions.ExpiredTokenException(
                _(
                    "This verification code has expired, "
                    "Please send a new code and try again."
                )
            )
        return token == self.token


class AbstractRegisteredUser(models.Model):
    user = models.OneToOneField(
        settings.AUTH_USER_MODEL,
        on_delete=models.CASCADE,
        related_name="registered_user",
        primary_key=True,
    )
    method = models.CharField(
        _("registration method"),
        help_text=_(
            "users can sign up in different ways, some methods are valid as "
            "indirect identity verification (eg: mobile phone SIM card in "
            "most countries)"
        ),
        max_length=64,
        blank=True,
        default="",
        choices=(
            REGISTRATION_METHOD_CHOICES
            if django.VERSION < (5, 0)
            # TODO: Remove when dropping support for Django 4.2
            # In Django 5.0+, choices are normalized at model definition,
            # creating a static list of tuples that doesn't update when registration
            # methods are dynamically registered or unregistered. Using a callable
            # ensures we always get the current choices from the registry.
            else get_registration_choices
        ),
    )
    is_verified = models.BooleanField(
        _("verified"),
        help_text=_(
            "whether the user has completed any identity "
            "verification process sucessfully"
        ),
        default=False,
    )
    modified = AutoLastModifiedField(_("Last verification change"), editable=True)
    _weak_verification_methods = {"", "email"}

    @property
    def is_identity_verified_strong(self):
        return self.is_verified and self.method not in self._weak_verification_methods

    class Meta:
        abstract = True
        verbose_name = _("Registration Information")
        verbose_name_plural = verbose_name

    @classmethod
    def unverify_inactive_users(cls):
        if not app_settings.UNVERIFY_INACTIVE_USERS:
            return
        # Exclude users who have unspecified, manual, or email
        # registration method because such users don't have an option
        # to re-verify. See https://github.com/openwisp/openwisp-radius/issues/517
        cls.objects.exclude(method__in=["", "manual", "email"]).filter(
            user__is_staff=False,
            user__last_login__lt=timezone.now()
            - timedelta(days=app_settings.UNVERIFY_INACTIVE_USERS),
        ).update(is_verified=False)

    @classmethod
    def delete_inactive_users(cls):
        if not app_settings.DELETE_INACTIVE_USERS:
            return
        cutoff_date = timezone.now() - timedelta(
            days=app_settings.DELETE_INACTIVE_USERS
        )
        User.objects.filter(
            Q(is_staff=False)
            & (
                Q(last_login__lt=cutoff_date)
                |
                # There could be manually created users which never logged in.
                # We use the "date_joined" field for such users.
                Q(last_login__isnull=True, date_joined__lt=cutoff_date)
            )
        ).delete()