[feature] Added mac address authentication for roaming users #490

Closes #490

docs/source/user/settings.rst

@@ -603,6 +603,31 @@ via the organization radius settings section of the admin interface.
 
 .. _openwisp_radius_needs_identity_verification:
 
+``OPENWISP_RADIUS_MAC_ADDR_ROAMING_ENABLED``
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+**Default**: ``False``
+
+Indicates whether MAC address roaming is supported.
+When this setting is enabled (i.e. ``True``),
+MAC address roaming is enabled for all organizations.
+
+**This setting can be overridden in individual organizations
+via the admin interface**, by going to *Organizations*
+then edit a specific organization and scroll down to
+*"Organization RADIUS settings"*, as shown in the screenshot below.
+
+.. image:: /images/mac-address-roaming.png
+   :alt: Organization MAC Address Roaming settings
+
+.. note::
+
+    We recommend using the override via the admin interface only when there
+    are special organizations which need a different configuration, otherwise,
+    if all the organization use the same configuration, we recommend
+    changing the global setting.
+
+
 ``OPENWISP_RADIUS_NEEDS_IDENTITY_VERIFICATION``
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

openwisp_radius/admin.py

@@ -552,6 +552,7 @@ class OrganizationRadiusSettingsInline(admin.StackedInline):
                     'registration_enabled',
                     'saml_registration_enabled',
                     'social_registration_enabled',
+                    'mac_addr_roaming_enabled',
                     'needs_identity_verification',
                     'first_name',
                     'last_name',

openwisp_radius/api/freeradius_views.py

@@ -1,6 +1,7 @@
 import ipaddress
 import logging
 import math
+import re
 
 import drf_link_header_pagination
 import swapper
@@ -37,6 +38,7 @@
 )
 from .utils import IDVerificationHelper
 
+RE_MAC_ADDR = re.compile(u'^{0}-{0}-{0}-{0}-{0}-{0}'.format(u'[a-f0-9]{2}'), re.I)
 _TOKEN_AUTH_FAILED = _('Token authentication failed')
 # Accounting-On and Accounting-Off are not implemented and
 # hence  ignored right now - may be implemented in the future
@@ -109,6 +111,31 @@ def _check_client_ip_and_return(self, request, uuid):
         logger.warning(message)
         raise AuthenticationFailed(message)
 
+    def _mac_address_roaming_auth(self, username, request):
+        calling_station_id = RE_MAC_ADDR.match(username)[0]
+        # Get the open session for the roaming user
+        open_session = (
+            RadiusAccounting.objects.select_related('organization__radius_settings')
+            .filter(calling_station_id=calling_station_id, stop_time=None)
+            .first()
+        )
+        if open_session and open_session.organization.radius_settings.get_setting(
+            'mac_addr_roaming_enabled'
+        ):
+            username = open_session.username
+            if hasattr(request.data, '_mutable'):
+                request.data._mutable = True
+            request.data['username'] = username
+            if hasattr(request.data, '_mutable'):
+                request.data._mutable = False
+            cache.set(
+                f'rt-{calling_station_id}',
+                str(open_session.organization_id),
+                86400,
+            )
+            request._mac_allowed = True
+        return username, request
+
     def _radius_token_authenticate(self, request):
         if request.data.get('status_type', None) in UNSUPPORTED_STATUS_TYPES:
             return
@@ -120,26 +147,8 @@ def _radius_token_authenticate(self, request):
         if cached_orgid:
             return self._check_client_ip_and_return(request, cached_orgid)
         else:
-
-            MACAUTH_ROAMING = True
-            mac_allowed = False
-            logger.warn(f'{username} and {len(username)}')
-            if MACAUTH_ROAMING and username and len(username) == 17:
-
-                open_session = RadiusAccounting.objects.filter(
-                    calling_station_id=username, stop_time=None
-                    # TODO: check ORG? HOW
-                ).first()
-
-                logger.warn(f'open_session: {open_session}')
-
-                if open_session:
-                    logger.warn(f'found open session: {open_session.__dict__}')
-                    username = open_session.username
-                    mac_allowed = True
-                    request.data['username'] = username
-            request._mac_allowed = mac_allowed
-
+            if username and RE_MAC_ADDR.search(username):
+                username, request = self._mac_address_roaming_auth(username, request)
             try:
                 radtoken = RadiusToken.objects.get(
                     user=auth_backend.get_users(username).first()
@@ -238,8 +247,6 @@ def post(self, request, *args, **kwargs):
         username = serializer.validated_data.get('username')
         password = serializer.validated_data.get('password')
 
-        logger.warn(f'DEBUG: {username}')
-
         user = self.get_user(request, username, password)
         if user and self.authenticate_user(request, user, password):
             data, status = self.get_replies(user, organization_id=request.auth)
@@ -249,40 +256,10 @@ def post(self, request, *args, **kwargs):
         else:
             return Response(None, status=200)
 
-    # def _radius_token_authenticate(self, request):
-    #     MACAUTH_ROAMING = True
-    #     mac_allowed = False
-    #     username = request.data.get('username')
-    #     password = request.data.get('password')
-    #     if MACAUTH_ROAMING and username == password:
-    #         open_session = RadiusAccounting.objects.filter(
-    #             calling_station_id=username, stop_time=None,
-    #             organization_id=request.auth
-    #         ).first()
-    #         if open_session:
-    #             logger.warn(f'found open session: {open_session.__dict__}')
-    #             username = open_session.username
-    #             mac_allowed = True
-    #             request.data['username'] = username
-    #             request.data['password'] = password
-    #     request._mac_allowed = mac_allowed
-    #     return super()._radius_token_authenticate(request)
-
     def get_user(self, request, username, password):
         """
         return user or ``None``
         """
-        # MACAUTH_ROAMING = True
-        # mac_allowed = False
-        # if MACAUTH_ROAMING and username == password:
-        #     open_session = RadiusAccounting.objects.filter(
-        #         calling_station_id=username, stop_time=None,
-        #         organization_id=request.auth
-        #     ).first()
-        #     if open_session:
-        #         logger.warn(f'found open session: {open_session.__dict__}')
-        #         username = open_session.username
-        #         mac_allowed = True
         conditions = self._get_user_query_conditions(request)
         try:
             user = auth_backend.get_users(username).filter(conditions)[0]

openwisp_radius/base/models.py

@@ -148,6 +148,9 @@
 _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'
 )
@@ -1202,6 +1205,14 @@ class AbstractOrganizationRadiusSettings(UUIDModel):
         verbose_name=_('SAML registration enabled'),
         fallback=app_settings.SAML_REGISTRATION_ENABLED,
     )
+    mac_addr_roaming_enabled = FallbackBooleanChoiceField(
+        null=True,
+        blank=True,
+        default=None,
+        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(
         null=True,
         blank=True,

openwisp_radius/migrations/0036_organizationradiussettings_mac_addr_roaming_enabled.py

@@ -0,0 +1,27 @@
+# Generated by Django 3.2.21 on 2023-10-11 12:06
+
+from django.db import migrations
+
+import openwisp_utils.fields
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('openwisp_radius', '0035_organizationradiussettings_sms_cooldown'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='organizationradiussettings',
+            name='mac_addr_roaming_enabled',
+            field=openwisp_utils.fields.FallbackBooleanChoiceField(
+                blank=True,
+                default=None,
+                fallback=False,
+                help_text='Whether the MAC address roaming should be enabled or not.',
+                null=True,
+                verbose_name='MAC address roaming enabled',
+            ),
+        ),
+    ]

openwisp_radius/settings.py

@@ -55,6 +55,7 @@ def get_default_password_reset_url(urls):
 SOCIAL_REGISTRATION_ENABLED = get_settings_value('SOCIAL_REGISTRATION_ENABLED', False)
 SAML_REGISTRATION_CONFIGURED = 'djangosaml2' in getattr(settings, 'INSTALLED_APPS', [])
 SAML_REGISTRATION_ENABLED = get_settings_value('SAML_REGISTRATION_ENABLED', False)
+MAC_ADDR_ROAMING_ENABLED = get_settings_value('MAC_ADDR_ROAMING_ENABLED', False)
 SAML_REGISTRATION_METHOD_LABEL = get_settings_value(
     'SAML_REGISTRATION_METHOD_LABEL', _('Single Sign-On (SAML)')
 )

openwisp_radius/tests/test_api/test_freeradius_api.py

@@ -1466,6 +1466,130 @@ def test_user_accounting_list_empty_diff_organization(self):
             self.assertEqual(len(response.json()), 0)
 
 
+class TestMacAddressRoaming(AcctMixin, ApiTokenMixin, BaseTestCase):
+    _test_email = '[email protected]'
+
+    def setUp(self):
+        cache.clear()
+        logging.disable(logging.WARNING)
+        super().setUp()
+
+    def test_mac_addr_roaming_authorize_view(self):
+        acct_post_data = self.acct_post_data
+        acct_post_data['username'] = 'tester'
+        acct_post_data['calling_station_id'] = '00-11-22-33-44-55'
+        self._get_org_user()
+        self._login_and_obtain_auth_token()
+
+        with self.subTest('Test user does not have an open session'):
+            response = self._authorize_user(
+                username=acct_post_data['calling_station_id'],
+                password=acct_post_data['calling_station_id'],
+            )
+            self.assertEqual(response.status_code, 403)
+
+        self._create_radius_accounting(**acct_post_data)
+
+        with self.subTest('Test mac address roaming is disabled'):
+            response = self._authorize_user(
+                username=acct_post_data['calling_station_id'],
+                password=acct_post_data['calling_station_id'],
+            )
+            self.assertEqual(response.status_code, 403)
+
+        OrganizationRadiusSettings.objects.update(mac_addr_roaming_enabled=True)
+        with self.subTest(
+            'Test user has open session and mac address roaming is enabled'
+        ):
+            response = self._authorize_user(
+                username=acct_post_data['calling_station_id'],
+                password=acct_post_data['calling_station_id'],
+            )
+            self.assertEqual(response.status_code, 200)
+            self.assertEqual(
+                response.data,
+                {
+                    'control:Auth-Type': 'Accept',
+                    'Session-Timeout': 10539,
+                    'ChilliSpot-Max-Total-Octets': 1487813647,
+                },
+            )
+        OrganizationRadiusSettings.objects.update(mac_addr_roaming_enabled=False)
+
+    def test_mac_addr_roaming_accounting_view(self):
+        acct_post_data = self.acct_post_data
+        acct_post_data['username'] = 'tester'
+        acct_post_data['calling_station_id'] = '00-11-22-33-44-55'
+        payload = acct_post_data.copy()
+        payload.update(
+            {
+                'unique_id': '119',
+                'username': payload['calling_station_id'],
+                'status_type': 'Start',
+                'nas_ip_address': '172.16.64.92',
+                'called_station_id': '66:55:44:33:22:11:hostname',
+            }
+        )
+
+        self._get_org_user()
+        self._login_and_obtain_auth_token()
+
+        with self.subTest('Test user does not have an open session'):
+            response = response = self.client.post(
+                self._acct_url,
+                data=json.dumps(payload),
+                content_type='application/json',
+            )
+            self.assertEqual(response.status_code, 403)
+
+        self._create_radius_accounting(update_time=now(), **acct_post_data)
+
+        with self.subTest('Test mac address roaming is disabled'):
+            response = response = self.client.post(
+                self._acct_url,
+                data=json.dumps(payload),
+                content_type='application/json',
+            )
+            self.assertEqual(response.status_code, 403)
+
+        OrganizationRadiusSettings.objects.update(mac_addr_roaming_enabled=True)
+        with self.subTest(
+            'Test user has open session and mac address roaming is enabled'
+        ):
+            response = response = self.client.post(
+                self._acct_url,
+                data=json.dumps(payload),
+                content_type='application/json',
+            )
+            self.assertEqual(response.status_code, 201)
+            self.assertEqual(
+                response.data,
+                None,
+            )
+            self.assertEqual(
+                RadiusAccounting.objects.filter(username='tester').count(), 2
+            )
+            self.assertEqual(
+                RadiusAccounting.objects.filter(
+                    username='tester',
+                    stop_time=None,
+                    nas_ip_address=payload['nas_ip_address'],
+                    called_station_id=payload['called_station_id'],
+                ).count(),
+                1,
+            )
+            self.assertEqual(
+                RadiusAccounting.objects.filter(
+                    username='tester',
+                    stop_time__isnull=False,
+                    nas_ip_address=acct_post_data['nas_ip_address'],
+                    called_station_id=acct_post_data['called_station_id'],
+                ).count(),
+                1,
+            )
+        OrganizationRadiusSettings.objects.update(mac_addr_roaming_enabled=False)
+
+
 class TestApiReject(ApiTokenMixin, BaseTestCase):
     @classmethod
     def setUpClass(cls):

tests/openwisp2/sample_radius/migrations/0028_alter_organizationradiussettings_allowed_mobile_prefixes_and_more.py

@@ -243,6 +243,18 @@ class Migration(migrations.Migration):
                 verbose_name='CoA Enabled',
             ),
         ),
+        migrations.AddField(
+            model_name='organizationradiussettings',
+            name='mac_addr_roaming_enabled',
+            field=openwisp_utils.fields.FallbackBooleanChoiceField(
+                blank=True,
+                default=None,
+                fallback=False,
+                help_text='Whether the MAC address roaming should be enabled or not.',
+                null=True,
+                verbose_name='MAC address roaming enabled',
+            ),
+        ),
         migrations.AddField(
             model_name="organizationradiussettings",
             name="sms_cooldown",