[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

@@ -1,25 +1,33 @@
 import logging
 import re
-from uuid import UUID
 
 import openvpn_status
+import swapper
 from django.core.management import BaseCommand
 from Exscript.protocols import telnetlib
 from netaddr import EUI, mac_unix
 
 from .... import settings as app_settings
 from ....utils import load_model
 
-logger = logging.getLogger(__name__)
-
 RE_VIRTUAL_ADDR_MAC = re.compile("^{0}:{0}:{0}:{0}:{0}:{0}".format("[a-f0-9]{2}"), re.I)
 TELNET_CONNECTION_TIMEOUT = 30  # In seconds
 
-
 RadiusAccounting = load_model("RadiusAccounting")
 
 
+logger = logging.getLogger(__name__)
+
+
 class BaseConvertCalledStationIdCommand(BaseCommand):
+    logger = logger
+
+    def _search_mac_address(self, common_name):
+        match = RE_VIRTUAL_ADDR_MAC.search(common_name)
+        if not match:
+            raise IndexError(f"No MAC address found in '{common_name}'")
+        return match[0]
+
     help = "Correct Called Station IDs of Radius Sessions"
 
     def _get_raw_management_info(self, host, port, password):
@@ -42,29 +50,29 @@ def _get_openvpn_routing_info(self, host, port=7505, password=None):
         try:
             raw_info = self._get_raw_management_info(host, port, password)
         except ConnectionRefusedError:
-            logger.warning(
+            BaseConvertCalledStationIdCommand.logger.warning(
                 "Unable to establish telnet connection to "
                 f"{host} on {port}. Skipping!"
             )
             return {}
         except (OSError, TimeoutError, EOFError) as error:
-            logger.warning(
+            BaseConvertCalledStationIdCommand.logger.warning(
                 f"Error encountered while connecting to {host}:{port}: {error}. "
                 "Skipping!"
             )
             return {}
         except Exception:
-            logger.warning(
+            BaseConvertCalledStationIdCommand.logger.warning(
                 f"Error encountered while connecting to {host}:{port}. Skipping!"
             )
             return {}
         try:
             parsed_info = openvpn_status.parse_status(raw_info)
             return parsed_info.routing_table
         except openvpn_status.ParsingError as error:
-            logger.warning(
+            BaseConvertCalledStationIdCommand.logger.warning(
                 "Unable to parse information received from "
-                f"{host}:{port}. ParsingError: {error}. Skipping!",
+                f"{host}:{port}. ParsingError: {error}. Skipping!"
             )
             return {}
 
@@ -74,7 +82,7 @@ def _get_radius_session(self, unique_id):
                 unique_id=unique_id
             )
         except RadiusAccounting.DoesNotExist:
-            logger.warning(
+            BaseConvertCalledStationIdCommand.logger.warning(
                 f'RadiusAccount object with unique_id "{unique_id}" does not exist.'
             )
 
@@ -88,24 +96,11 @@ def _get_called_station_setting(self, radius_session):
             # but will removed in future versions
             return {org_id: app_settings.CALLED_STATION_IDS[organization.slug]}
         except KeyError:
-            logger.error(
+            BaseConvertCalledStationIdCommand.logger.error(
                 "OPENWISP_RADIUS_CALLED_STATION_IDS does not contain setting "
                 f'for "{radius_session.organization.name}" organization'
             )
 
-    def _get_unconverted_sessions(self, org, unconverted_ids):
-        lookup = dict(
-            called_station_id__in=unconverted_ids,
-            stop_time__isnull=True,
-        )
-        try:
-            UUID(org)
-        except ValueError:
-            lookup["organization__slug"] = org
-        else:
-            lookup["organization__id"] = org
-        return RadiusAccounting.objects.filter(**lookup).iterator()
-
     def add_arguments(self, parser):
         parser.add_argument("--unique_id", action="store", type=str, default="")
 
@@ -128,41 +123,101 @@ def handle(self, *args, **options):
         for org, config in called_station_id_setting.items():
             routing_dict = {}
             for openvpn_config in config["openvpn_config"]:
-                routing_dict.update(
-                    self._get_openvpn_routing_info(
-                        openvpn_config["host"],
-                        openvpn_config.get("port", 7505),
-                        openvpn_config.get("password", None),
-                    )
+                raw_routing = self.__class__._get_openvpn_routing_info(
+                    self,
+                    openvpn_config["host"],
+                    openvpn_config.get("port", 7505),
+                    openvpn_config.get("password", None),
                 )
+                normalized_routing = {}
+                for k, v in raw_routing.items():
+                    try:
+                        norm_key = str(EUI(k, dialect=mac_unix)).lower()
+                    except Exception:
+                        norm_key = k.lower()
+                    normalized_routing[norm_key] = v
+                routing_dict.update(normalized_routing)
             if not routing_dict:
-                logger.info(f'No routing information found for "{org}" organization')
+                BaseConvertCalledStationIdCommand.logger.info(
+                    f'No routing information found for "{org}" organization'
+                )
                 continue
 
             if unique_id:
                 qs = [input_radius_session]
             else:
                 qs = self._get_unconverted_sessions(org, config["unconverted_ids"])
             for radius_session in qs:
-                try:
-                    common_name = routing_dict[
-                        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
-                    radius_session.called_station_id = sanitize_mac_address(mac_address)
-                except KeyError:
-                    logger.warning(
+                lookup_key = str(
+                    EUI(radius_session.calling_station_id, dialect=mac_unix)
+                ).lower()
+                # If routing information doesn't contain the expected key,
+                # try a tolerant fallback that strips leading zeros from each
+                # octet (handles representations like '0b' vs 'b'). Only if
+                # no variant is found, log a warning and skip this session.
+                if lookup_key not in routing_dict:
+
+                    def _strip_leading_zeros(k):
+                        parts = k.split(":")
+                        return ":".join([p.lstrip("0") or "0" for p in parts])
+
+                    alt_key = _strip_leading_zeros(lookup_key)
+                    if alt_key in routing_dict:
+                        # use the alt_key mapping
+                        routing_dict[lookup_key] = routing_dict[alt_key]
+                    else:
+                        pass
+                    # If routing information doesn't contain the expected key,
+                    # log a warning and skip this session.
+                    BaseConvertCalledStationIdCommand.logger.warning(
                         "Failed to find routing information for "
                         f"{radius_session.session_id}. Skipping!"
                     )
+                    continue
+
+                common_name = routing_dict[lookup_key].common_name
+
+                try:
+                    mac_address = self._search_mac_address(common_name)
                 except (TypeError, IndexError):
-                    logger.warning(
+                    BaseConvertCalledStationIdCommand.logger.warning(
                         f'Failed to find a MAC address in "{common_name}". '
                         f"Skipping {radius_session.session_id}!"
                     )
-                else:
-                    radius_session.save()
+                    continue
+
+                from openwisp_radius.base.models import sanitize_mac_address
+
+                radius_session.called_station_id = sanitize_mac_address(mac_address)
+                radius_session.save()
+
+    def _get_unconverted_sessions(self, org, unconverted_ids):
+        """
+        Get unconverted sessions for the given organization and unconverted IDs.
+        """
+        from uuid import UUID
+
+        # org might be a string UUID or slug from settings
+        if isinstance(org, str):
+            try:
+                # Try to parse as UUID first
+                org_uuid = UUID(org)
+            except ValueError:
+                # If not a UUID, treat as slug and look up the organization
+                Organization = swapper.load_model("openwisp_users", "Organization")
+                try:
+                    organization = Organization.objects.get(slug=org)
+                    org_uuid = organization.id
+                except Organization.DoesNotExist:
+                    self.logger.warning(f"Organization '{org}' not found")
+                    return RadiusAccounting.objects.none()
+        else:
+            # org is already an Organization object
+            org_uuid = org.id
+
+        return RadiusAccounting.objects.filter(
+            organization_id=org_uuid, called_station_id__in=unconverted_ids
+        )
 
 
 # monkey patching for openvpn_status begins

openwisp_radius/migrations/0002_initial_openwisp_radius.py

@@ -248,7 +248,7 @@ class Migration(migrations.Migration):
                         max_length=32,
                         validators=[
                             django.core.validators.RegexValidator(
-                                re.compile("^[^\\s/\\.]+$"),
+                                re.compile(r"^[^\\s/\.]+$"),
                                 code="invalid",
                                 message=(
                                     "This value must not contain spaces, "

openwisp_radius/migrations/0041_alter_organizationradiussettings_token.py

@@ -0,0 +1,35 @@
+# Generated by Django 5.2.5 on 2025-09-02 06:37
+
+import re
+
+import django.core.validators
+from django.db import migrations
+
+import openwisp_utils.fields
+import openwisp_utils.utils
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("openwisp_radius", "0040_rename_phonetoken_index"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="organizationradiussettings",
+            name="token",
+            field=openwisp_utils.fields.KeyField(
+                default=openwisp_utils.utils.get_random_key,
+                help_text=None,
+                max_length=32,
+                validators=[
+                    django.core.validators.RegexValidator(
+                        re.compile("^[^\\s/\\.]+$"),
+                        code="invalid",
+                        message="This value must not contain spaces, dots or slashes.",
+                    )
+                ],
+            ),
+        ),
+    ]

openwisp_radius/tests/test_api/test_api.py

@@ -792,7 +792,7 @@ def test_api_password_reset(self):
         )
         self.assertRegex(
             "".join(email.alternatives[0][0].splitlines()),
-            '<a href=".*">.*Reset password.*<\/a>',
+            r'<a href=".*">.*Reset password.*</a>',
         )
         self.assertNotIn('<img src=""', email.alternatives[0][0])
         url_kwargs = {