[feature] Sanitize MAC addresses in called_station_id field

Author: Eeshu-Yadav

openwisp_radius/api/freeradius_views.py

@@ -28,6 +28,7 @@
 
 from .. import registration
 from .. import settings as app_settings
+from ..base.models import sanitize_mac_address
 from ..counters.base import BaseCounter
 from ..counters.exceptions import MaxQuotaReached
 from ..signals import radius_accounting_success
@@ -372,8 +373,8 @@ def _check_simultaneous_use(
         # of the same client on the same NAS.
         if called_station_id and calling_station_id:
             open_sessions = open_sessions.exclude(
-                called_station_id=called_station_id,
-                calling_station_id=calling_station_id,
+                called_station_id=sanitize_mac_address(called_station_id),
+                calling_station_id=sanitize_mac_address(calling_station_id),
             )
         open_sessions = open_sessions.count()
         if open_sessions >= max_simultaneous:

openwisp_radius/base/models.py

@@ -31,49 +31,6 @@
 from phonenumber_field.modelfields import PhoneNumberField
 from private_storage.fields import PrivateFileField
 
-def sanitize_mac_address(mac):
-    """
-    Sanitize a MAC address string to the colon-separated lowercase format.
-    If the input is not a valid MAC address, return it unchanged.
-    
-    Handles various MAC address formats:
-    - 00:1A:2B:3C:4D:5E -> 00:1a:2b:3c:4d:5e
-    - 00-1A-2B-3C-4D-5E -> 00:1a:2b:3c:4d:5e
-    - 001A.2B3C.4D5E -> 00:1a:2b:3c:4d:5e
-    - 001A2B3C4D5E -> 00:1a:2b:3c:4d:5e
-    """
-    # Return empty string or non-string input as is
-    if not mac or not isinstance(mac, str):
-        return mac
-        
-    # Check if it's an IP address (IPv4) - preserve unchanged
-    if re.match(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$', mac):
-        return mac
-    
-    # Try to extract MAC address from the string
-    # Look for MAC address patterns, including those with additional text
-    mac_patterns = [
-        r'([0-9A-Fa-f]{2}[:-]){5}[0-9A-Fa-f]{2}',  # Standard MAC with : or -
-        r'([0-9A-Fa-f]{4}\.){2}[0-9A-Fa-f]{4}',    # Cisco format
-        r'[0-9A-Fa-f]{12}'                          # No separators
-    ]
-    
-    for pattern in mac_patterns:
-        match = re.search(pattern, mac)
-        if match:
-            mac_candidate = match.group(0)
-            try:
-                # Try to parse as EUI to validate
-                eui = EUI(mac_candidate)
-                # Return sanitized MAC address in colon-separated lowercase format
-                return ':'.join(['%02x' % x for x in eui.words]).lower()
-            except (AddrFormatError, ValueError, TypeError):
-                continue
-    
-    # If no valid MAC found, return original string unchanged
-    return mac
-
-import swapper
 from openwisp_radius.registration import (
     REGISTRATION_METHOD_CHOICES,
     get_registration_choices,
@@ -223,6 +180,53 @@ def sanitize_mac_address(mac):
 OPTIONAL_SETTINGS = app_settings.OPTIONAL_REGISTRATION_FIELDS
 
 
+def sanitize_mac_address(mac):
+    """
+    Sanitize a MAC address string to the colon-separated lowercase format.
+    If the input is not a valid MAC address, return it unchanged.
+
+    Handles various MAC address formats:
+    - 00:1A:2B:3C:4D:5E -> 00:1a:2b:3c:4d:5e
+    - 00-1A-2B-3C-4D-5E -> 00:1a:2b:3c:4d:5e
+    - 001A.2B3C.4D5E -> 00:1a:2b:3c:4d:5e
+    - 001A2B3C4D5E -> 00:1a:2b:3c:4d:5e
+    """
+    if not mac or not isinstance(mac, str):
+        return mac
+    try:
+        ipaddress.ip_address(mac)
+        return mac
+    except ValueError:
+        pass
+    mac_patterns = [
+        r"([0-9A-Fa-f]{2}[:-]){5}[0-9A-Fa-f]{2}",
+        r"([0-9A-Fa-f]{4}\.){2}[0-9A-Fa-f]{4}",
+        r"[0-9A-Fa-f]{12}",
+    ]
+    for pattern in mac_patterns:
+        if re.fullmatch(pattern, mac):
+            try:
+                eui = EUI(mac)
+                return ":".join([f"{x:02x}" for x in eui.words]).lower()
+            except (AddrFormatError, ValueError, TypeError):
+                continue
+    return mac
+
+
+class SanitizedMacCharField(models.CharField):
+    """
+    A CharField that automatically sanitizes MAC addresses
+    to the canonical lowercase colon-separated format on save.
+    """
+
+    def pre_save(self, model_instance, add):
+        value = getattr(model_instance, self.attname)
+        if value:
+            value = sanitize_mac_address(value)
+            setattr(model_instance, self.attname, value)
+        return value
+
+
 class AutoUsernameMixin(object):
     def clean(self):
         """
@@ -495,15 +499,15 @@ class AbstractRadiusAccounting(OrgMixin, models.Model):
         null=True,
         blank=True,
     )
-    called_station_id = models.CharField(
+    called_station_id = SanitizedMacCharField(
         verbose_name=_("called station ID"),
         max_length=253,
         db_column="calledstationid",
         db_index=True,
         blank=True,
         null=True,
     )
-    calling_station_id = models.CharField(
+    calling_station_id = SanitizedMacCharField(
         verbose_name=_("calling station ID"),
         max_length=253,
         db_column="callingstationid",
@@ -573,8 +577,6 @@ class AbstractRadiusAccounting(OrgMixin, models.Model):
     )
 
     def save(self, *args, **kwargs):
-        if self.called_station_id:
-            self.called_station_id = sanitize_mac_address(self.called_station_id)
         if not self.start_time:
             self.start_time = now()
         super(AbstractRadiusAccounting, self).save(*args, **kwargs)
@@ -624,8 +626,9 @@ def _close_stale_sessions_on_nas_boot(cls, called_station_id):
         """
         if not called_station_id:
             return 0
+        sanitized_called_station_id = sanitize_mac_address(called_station_id)
         stale_sessions = cls.objects.filter(
-            called_station_id=called_station_id,
+            called_station_id=sanitized_called_station_id,
             stop_time__isnull=True,
         )
         closed_count = stale_sessions.update(
@@ -877,14 +880,14 @@ class AbstractRadiusPostAuth(OrgMixin, UUIDModel):
         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(
+    called_station_id = SanitizedMacCharField(
         verbose_name=_("called station ID"),
         max_length=253,
         db_column="calledstationid",
         blank=True,
         null=True,
     )
-    calling_station_id = models.CharField(
+    calling_station_id = SanitizedMacCharField(
         verbose_name=_("calling station ID"),
         max_length=253,
         db_column="callingstationid",