[models] Sanitize MAC addresses in called_station_id field #624

This change implements MAC address sanitization in the called_station_id
field of RadiusAccounting to ensure consistent formatting across different
input formats. MAC addresses are now automatically converted to lowercase
colon-separated format (aa:bb:cc:dd:ee:ff) while preserving non-MAC values
unchanged for RFC compliance.

Changes:
- Added sanitize_mac_address function in base/models.py using netaddr.EUI
- Integrated automatic MAC sanitization in AbstractRadiusAccounting.save()
- Updated test imports to use sanitize_mac_address from base.models
- Supports multiple MAC formats: colon, dash, dot, and no separators

Fixes #624

Author: Eeshu-Yadav

openwisp_radius/base/models.py

@@ -28,49 +28,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,
@@ -104,7 +61,6 @@ def sanitize_mac_address(mac):
     prefix_generate_users,
     validate_csvfile,
 )
-from ..mac_utils import sanitize_mac_address
 from .validators import ipv6_network_validator, password_reset_url_validator
 
 logger = logging.getLogger(__name__)
@@ -220,6 +176,49 @@ 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
+    """
+    # 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
+
+
 class AutoUsernameMixin(object):
     def clean(self):
         """

openwisp_radius/management/commands/base/convert_called_station_id.py

@@ -149,7 +149,8 @@ def handle(self, *args, **options):
                         str(EUI(radius_session.calling_station_id, dialect=mac_unix))
                     ].common_name
                     mac_address = RE_VIRTUAL_ADDR_MAC.search(common_name)[0]
-                    from openwisp_radius.mac_utils import sanitize_mac_address
+                    from openwisp_radius.base.models import sanitize_mac_address
+
                     radius_session.called_station_id = sanitize_mac_address(mac_address)
                 except KeyError:
                     logger.warning(

openwisp_radius/tests/test_api/test_freeradius_api.py

@@ -20,8 +20,8 @@
 from ... import registration
 from ... import settings as app_settings
 from ...api.freeradius_views import logger as freeradius_api_logger
-from ...counters.exceptions import MaxQuotaReached, SkipCheck
 from ...base.models import sanitize_mac_address
+from ...counters.exceptions import MaxQuotaReached, SkipCheck
 from ...signals import radius_accounting_success
 from ...utils import load_model
 from ..mixins import ApiTokenMixin, BaseTestCase, BaseTransactionTestCase
@@ -406,7 +406,9 @@ def assertAcctData(self, ra, data):
             if key == "called_station_id":
                 # Both values should be sanitized for comparison
                 ra_value = sanitize_mac_address(ra_value) if ra_value else ra_value
-                data_value = sanitize_mac_address(data_value) if data_value else data_value
+                data_value = (
+                    sanitize_mac_address(data_value) if data_value else data_value
+                )
             _type = type(ra_value)
             if _type != type(data_value):
                 data_value = _type(data_value)
@@ -1853,7 +1855,9 @@ def test_mac_addr_roaming_accounting_view(self):
                     username="tester",
                     stop_time__isnull=False,
                     nas_ip_address=acct_post_data["nas_ip_address"],
-                    called_station_id=sanitize_mac_address(acct_post_data["called_station_id"]),
+                    called_station_id=sanitize_mac_address(
+                        acct_post_data["called_station_id"]
+                    ),
                 ).count(),
                 1,
             )
@@ -2225,42 +2229,6 @@ def test_ip_from_radsetting_not_exist(self):
             self.assertEqual(response.status_code, 403)
             self.assertEqual(response.data["detail"], self.fail_msg)
 
-    def test_ip_from_radsetting_valid(self):
-        with mock.patch(self.freeradius_hosts_path, []):
-            radsetting = OrganizationRadiusSettings.objects.get(
-                organization=self._get_org()
-            )
-        radsetting.freeradius_allowed_hosts = "127.0.0.1"
-        radsetting.save()
-        response = self.client.post(reverse("radius:authorize"), self.params)
-        self.assertEqual(response.status_code, 200)
-        self.assertEqual(response.data, _AUTH_TYPE_ACCEPT_RESPONSE)
-
-    def test_ip_from_setting_valid(self):
-        response = self.client.post(reverse("radius:authorize"), self.params)
-        self.assertEqual(response.status_code, 200)
-        self.assertEqual(response.data, _AUTH_TYPE_ACCEPT_RESPONSE)
-
-    @capture_any_output()
-    def test_ip_from_radsetting_not_exist(self):
-        org2 = self._create_org(**{"name": "test", "slug": "test"})
-        user = self._create_user(username="org2-tester", email="tester@org2.com")
-        self._create_org_user(**{"organization": org2, "user": user})
-        params = self.params.copy()
-        params["username"] = "org2-tester"
-        response = self.client.post(
-            reverse("radius:user_auth_token", args=[org2.slug]), params
-        )
-        self.assertEqual(response.status_code, 200)
-        with self.subTest("FREERADIUS_ALLOWED_HOSTS is 127.0.0.1"):
-            response = self.client.post(reverse("radius:authorize"), params)
-            self.assertEqual(response.status_code, 200)
-            self.assertEqual(response.data, _AUTH_TYPE_ACCEPT_RESPONSE)
-        with self.subTest("Empty Settings"), mock.patch(self.freeradius_hosts_path, []):
-            response = self.client.post(reverse("radius:authorize"), params)
-            self.assertEqual(response.status_code, 403)
-            self.assertEqual(response.data["detail"], self.fail_msg)
-
 
 class TestTransactionClientIpApi(
     TestClientIpApiMixin, ApiTokenMixin, BaseTransactionTestCase