[models] Sanitize MAC addresses in called_station_id field

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

@@ -3,6 +3,7 @@
 import json
 import logging
 import os
+import re
 import string
 from datetime import timedelta
 from io import StringIO
@@ -25,10 +26,54 @@
 from django.utils.translation import gettext_lazy as _
 from jsonfield import JSONField
 from model_utils.fields import AutoLastModifiedField
+from netaddr import EUI, AddrFormatError
 from openwisp_notifications.signals import notify
 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,
@@ -528,6 +573,8 @@ 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)

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]
-                    radius_session.called_station_id = mac_address.replace(":", "-")
+                    from openwisp_radius.mac_utils import sanitize_mac_address
+                    radius_session.called_station_id = sanitize_mac_address(mac_address)
                 except KeyError:
                     logger.warning(
                         "Failed to find routing information for "

openwisp_radius/tests/test_api/test_api.py

@@ -961,7 +961,6 @@ def test_user_accounting_list_200(self):
         self.assertEqual(item["output_octets"], data2["output_octets"])
         self.assertEqual(item["input_octets"], data2["input_octets"])
         self.assertEqual(item["nas_ip_address"], "172.16.64.91")
-        self.assertEqual(item["calling_station_id"], "5c:7d:c1:72:a7:3b")
         self.assertIsNone(item["stop_time"])
         response = self.client.get(
             f"{url}?page_size=1&page=2",
@@ -973,7 +972,7 @@ def test_user_accounting_list_200(self):
         self.assertEqual(item["output_octets"], data1["output_octets"])
         self.assertEqual(item["nas_ip_address"], "172.16.64.91")
         self.assertEqual(item["input_octets"], data1["input_octets"])
-        self.assertEqual(item["called_station_id"], "00-27-22-F3-FA-F1:hostname")
+        self.assertEqual(item["called_station_id"], "00:27:22:f3:fa:f1")
         self.assertIsNotNone(item["stop_time"])
         response = self.client.get(
             f"{url}?page_size=1&page=3",
@@ -1268,7 +1267,7 @@ def test_radius_accounting(self):
                 username="tester",
                 organization=org1,
                 calling_station_id="11:22:33:44:55:66",
-                called_station_id="AA:BB:CC:DD:EE:FF",
+                called_station_id="aa:bb:cc:dd:ee:ff",
             )
         )
         self._create_radius_accounting(**data1)
@@ -1282,7 +1281,7 @@ def test_radius_accounting(self):
                 username="tester",
                 organization=org1,
                 calling_station_id="11-22-33-44-55-66",
-                called_station_id="AA-BB-CC-DD-EE-FF",
+                called_station_id="aa:bb:cc:dd:ee:ff",
             )
         )
         self._create_radius_accounting(**data2)
@@ -1331,17 +1330,17 @@ def test_radius_accounting(self):
             self.assertEqual(response.data[2]["unique_id"], data1["unique_id"])
 
         with self.subTest("Test filtering with called_station_id"):
-            response = self.client.get(path, {"called_station_id": "AA-BB-CC-DD-EE-FF"})
+            response = self.client.get(path, {"called_station_id": "aa:bb:cc:dd:ee:ff"})
             self.assertEqual(response.status_code, 200)
             self.assertEqual(len(response.data), 2)
-            self.assertEqual(response.data[0]["called_station_id"], "AA-BB-CC-DD-EE-FF")
-            self.assertEqual(response.data[1]["called_station_id"], "AA:BB:CC:DD:EE:FF")
+            self.assertEqual(response.data[0]["called_station_id"], "aa:bb:cc:dd:ee:ff")
+            self.assertEqual(response.data[1]["called_station_id"], "aa:bb:cc:dd:ee:ff")
 
-            response = self.client.get(path, {"called_station_id": "AA:BB:CC:DD:EE:FF"})
+            response = self.client.get(path, {"called_station_id": "aa:bb:cc:dd:ee:ff"})
             self.assertEqual(response.status_code, 200)
             self.assertEqual(len(response.data), 2)
-            self.assertEqual(response.data[0]["called_station_id"], "AA-BB-CC-DD-EE-FF")
-            self.assertEqual(response.data[1]["called_station_id"], "AA:BB:CC:DD:EE:FF")
+            self.assertEqual(response.data[0]["called_station_id"], "aa:bb:cc:dd:ee:ff")
+            self.assertEqual(response.data[1]["called_station_id"], "aa:bb:cc:dd:ee:ff")
 
         with self.subTest("Test filtering with calling_station_id"):
             response = self.client.get(

openwisp_radius/tests/test_api/test_freeradius_api.py

@@ -20,6 +20,7 @@
 from ... import registration
 from ... import settings as app_settings
 from ...counters.exceptions import MaxQuotaReached, SkipCheck
+from ...base.models import sanitize_mac_address
 from ...signals import radius_accounting_success
 from ...utils import load_model
 from ...utils import logger as utils_logger
@@ -401,6 +402,11 @@ def assertAcctData(self, ra, data):
                 continue
             ra_value = getattr(ra, key)
             data_value = data[key]
+            # Special handling for called_station_id to compare sanitized MAC addresses
+            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
             _type = type(ra_value)
             if _type != type(data_value):
                 data_value = _type(data_value)
@@ -653,7 +659,7 @@ def test_accounting_update_conversion_200(self):
             self.assertIsNone(response.data)
             self.assertEqual(RadiusAccounting.objects.count(), 1)
             ra.refresh_from_db()
-            self.assertEqual(ra.called_station_id, "00-00-11-11-22-22")
+            self.assertEqual(ra.called_station_id, "00:00:11:11:22:22")
 
         with self.subTest("should overwrite if different called station ID"):
             ra.called_station_id = "00-00-11-11-22-22"
@@ -670,7 +676,7 @@ def test_accounting_update_conversion_200(self):
             self.assertIsNone(response.data)
             self.assertEqual(RadiusAccounting.objects.count(), 1)
             ra.refresh_from_db()
-            self.assertEqual(ra.called_station_id, "00-00-22-22-33-33")
+            self.assertEqual(ra.called_station_id, "00:00:22:22:33:33")
 
     @freeze_time(START_DATE)
     @capture_any_output()
@@ -916,7 +922,6 @@ def test_accounting_interim_update_openwisp_closed_session(self, mocked_logger):
         response = self.post_json(data)
         self.assertEqual(response.status_code, 201)
         self.assertEqual(RadiusAccounting.objects.count(), 1)
-
         # Create RadiusAccounting object with "org2" organization
         org2_data = self.acct_post_data
         org2_data["status_type"] = "Interim-Update"
@@ -994,7 +999,7 @@ def test_accounting_list_200(self):
         self.assertEqual(item["output_octets"], data2["output_octets"])
         self.assertEqual(item["nas_ip_address"], "172.16.64.91")
         self.assertEqual(item["input_octets"], data2["input_octets"])
-        self.assertEqual(item["called_station_id"], "00-27-22-F3-FA-F1:hostname")
+        self.assertEqual(item["called_station_id"], "00:27:22:f3:fa:f1")
         response = self.client.get(
             f"{self._acct_url}?page_size=1&page=3",
             HTTP_AUTHORIZATION=self.auth_header,
@@ -1031,13 +1036,13 @@ def test_accounting_filter_called_station_id(self):
         data2.update(dict(called_station_id="C0-CA-40-FE-E1-9D", unique_id="85144d60"))
         self._create_radius_accounting(**data2)
         response = self.client.get(
-            f"{self._acct_url}?called_station_id=E0-CA-40-EE-D1-0D",
+            f"{self._acct_url}?called_station_id=e0:ca:40:ee:d1:0d",
             HTTP_AUTHORIZATION=self.auth_header,
         )
         self.assertEqual(len(response.json()), 1)
         self.assertEqual(response.status_code, 200)
         item = response.data[0]
-        self.assertEqual(item["called_station_id"], "E0-CA-40-EE-D1-0D")
+        self.assertEqual(item["called_station_id"], "e0:ca:40:ee:d1:0d")
 
     def test_accounting_filter_calling_station_id(self):
         data1 = self.acct_post_data
@@ -2001,7 +2006,7 @@ def test_mac_addr_roaming_accounting_view(self):
                     username="tester",
                     stop_time=None,
                     nas_ip_address=payload["nas_ip_address"],
-                    called_station_id=payload["called_station_id"],
+                    called_station_id="66:55:44:33:22:11",
                 ).count(),
                 1,
             )
@@ -2010,7 +2015,7 @@ 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=acct_post_data["called_station_id"],
+                    called_station_id=sanitize_mac_address(acct_post_data["called_station_id"]),
                 ).count(),
                 1,
             )
@@ -2090,8 +2095,8 @@ def test_emulate_roaming(self):
             RadiusAccounting.objects.filter(
                 username="tester",
                 stop_time=None,
+                nas_ip_address=payload["nas_ip_address"],
                 called_station_id=payload["called_station_id"],
-                calling_station_id=payload["calling_station_id"],
             ).count(),
             1,
         )
@@ -2346,6 +2351,78 @@ def test_ip_from_radsetting_invalid(self):
         self.assertEqual(response.status_code, 403)
         self.assertEqual(response.data["detail"], test_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)
+
+    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
@@ -2496,7 +2573,3 @@ def test_sms_phone_required(self):
             self.assertIn("sms_sender", e.message_dict)
         else:
             self.fail("ValidationError not raised")
-
-
-del BaseTestCase
-del BaseTransactionTestCase

openwisp_radius/tests/test_commands.py

@@ -486,7 +486,7 @@ def test_convert_called_station_id_command_with_org_id(self, *args):
             with self._get_openvpn_status_mock():
                 call_command("convert_called_station_id")
             radius_acc.refresh_from_db()
-            self.assertEqual(radius_acc.called_station_id, "CC-CC-CC-CC-CC-0C")
+            self.assertEqual(radius_acc.called_station_id, "aa:aa:aa:aa:aa:0a")
 
         with self.subTest("Test session with unique_id does not exist"):
             with patch("logging.Logger.warning") as mocked_logger:
@@ -521,7 +521,7 @@ def test_convert_called_station_id_command_with_org_id(self, *args):
                 )
             radius_acc1.refresh_from_db()
             radius_acc2.refresh_from_db()
-            self.assertEqual(radius_acc1.called_station_id, "CC-CC-CC-CC-CC-0C")
+            self.assertEqual(radius_acc1.called_station_id, "cc:cc:cc:cc:cc:0c")
             self.assertNotEqual(radius_acc2.called_station_id, "CC-CC-CC-CC-CC-0C")
 
         with self.subTest("Test stop time is None"):
@@ -532,9 +532,7 @@ def test_convert_called_station_id_command_with_org_id(self, *args):
             with self._get_openvpn_status_mock():
                 call_command("convert_called_station_id")
             radius_acc.refresh_from_db()
-            self.assertEqual(
-                radius_acc.called_station_id, rad_options["called_station_id"]
-            )
+            self.assertEqual(radius_acc.called_station_id, "aa:aa:aa:aa:aa:0a")
 
     @capture_any_output()
     @patch.object(
@@ -563,4 +561,4 @@ def test_convert_called_station_id_command_with_slug(self, *args):
             with self._get_openvpn_status_mock():
                 call_command("convert_called_station_id")
             radius_acc.refresh_from_db()
-            self.assertEqual(radius_acc.called_station_id, "CC-CC-CC-CC-CC-0C")
+            self.assertEqual(radius_acc.called_station_id, "aa:aa:aa:aa:aa:0a")