[fix] Handle importing UTF-16 encoded CSV files

CHANGES.rst

@@ -8,8 +8,8 @@ Features
 ~~~~~~~~
 
 - Added integration with `OpenWISP Monitoring
-  <https://openwisp.io/docs/stable/radius/user/radius_monitoring.html>`_ to
-  collect and visualize metrics for user-#s and RADIUS traffic.
+  <https://openwisp.io/docs/stable/radius/user/radius_monitoring.html>`_
+  to collect and visualize metrics for user-#s and RADIUS traffic.
 - Added support for `Change of Authorization (CoA)
   <https://openwisp.io/docs/stable/radius/user/change_of_authorization.html>`_.
 - Added `MonthlyTrafficCounter

README.rst

@@ -46,8 +46,8 @@ verification
 `generation of new users for events
 <https://openwisp.io/docs/stable/radius/user/generating_users.html>`_,
 `social login
-<https://openwisp.io/docs/stable/radius/user/social_login.html>`_, and much
-more.
+<https://openwisp.io/docs/stable/radius/user/social_login.html>`_, and
+much more.
 
 It can be used as a standalone application or integrated with the rest of
 `OpenWISP <https://openwisp.org>`_. It can also be used as a `base system

docs/developer/extending.rst

@@ -206,9 +206,7 @@ Once you have created the models, add the following to your
     OPENWISP_RADIUS_RADIUSGROUP_MODEL = "myradius.RadiusGroup"
     OPENWISP_RADIUS_RADIUSTOKEN_MODEL = "myradius.RadiusToken"
     OPENWISP_RADIUS_PHONETOKEN_MODEL = "myradius.PhoneToken"
-    OPENWISP_RADIUS_ORGANIZATIONRADIUSSETTINGS_MODEL = (
-        "myradius.OrganizationRadiusSettings"
-    )
+    OPENWISP_RADIUS_ORGANIZATIONRADIUSSETTINGS_MODEL = "myradius.OrganizationRadiusSettings"
     OPENWISP_RADIUS_REGISTEREDUSER_MODEL = "myradius.RegisteredUser"
 
     # You will need to change AUTH_USER_MODEL if you are extending openwisp_users
@@ -315,9 +313,7 @@ resort to monkey patching, you can proceed as follows:
     RadiusGroupCheck = load_model("openwisp_radius", "RadiusGroupCheck")
     RadiusGroupReply = load_model("openwisp_radius", "RadiusGroupReply")
     RadiusUserGroup = load_model("openwisp_radius", "RadiusUserGroup")
-    OrganizationRadiusSettings = load_model(
-        "openwisp_radius", "OrganizationRadiusSettings"
-    )
+    OrganizationRadiusSettings = load_model("openwisp_radius", "OrganizationRadiusSettings")
     User = get_user_model()
 
     admin.site.unregister(RadiusCheck)

docs/user/settings.rst

@@ -539,9 +539,7 @@ means that the global setting specified in ``settings.py`` will be used.
 
 .. code-block:: python
 
-    {
-        "__all__": "https://{site}/{organization}/password/reset/confirm/{uid}/{token}"
-    }
+    {"__all__": "https://{site}/{organization}/password/reset/confirm/{uid}/{token}"}
 
 A dictionary representing the frontend URLs through which end users can
 complete the password reset operation.
@@ -871,9 +869,7 @@ type in the API response for ``ChilliSpot-Max-Input-Octets`` attribute:
 
 .. code-block:: python
 
-    OPENWISP_RADIUS_RADIUS_ATTRIBUTES_TYPE_MAP = {
-        "ChilliSpot-Max-Input-Octets": "bytes"
-    }
+    OPENWISP_RADIUS_RADIUS_ATTRIBUTES_TYPE_MAP = {"ChilliSpot-Max-Input-Octets": "bytes"}
 
 .. _radius_social_login_settings:
 

openwisp_radius/api/views.py

@@ -155,9 +155,9 @@ def get(self, request, *args, **kwargs):
         if radbatch.strategy == 'prefix':
             pdf = generate_pdf(radbatch.pk)
             response = HttpResponse(content_type='application/pdf')
-            response[
-                'Content-Disposition'
-            ] = f'attachment; filename="{radbatch.name}.pdf"'
+            response['Content-Disposition'] = (
+                f'attachment; filename="{radbatch.name}.pdf"'
+            )
             response.write(pdf)
             return response
         else:

openwisp_radius/base/forms.py

@@ -46,7 +46,9 @@ def clean(self):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         if 'csvfile' in self.fields:
-            docs_link = 'https://openwisp.io/docs/stable/radius/user/importing_users.html'
+            docs_link = (
+                'https://openwisp.io/docs/stable/radius/user/importing_users.html'
+            )
             help_text = f"Refer to the <b><u><a href='{docs_link}'>docs</a></u></b> \
                 for more details on importing users from a CSV"
             self.fields['csvfile'].help_text = help_text

openwisp_radius/base/models.py

@@ -49,6 +49,7 @@
     SmsMessage,
     find_available_username,
     generate_sms_token,
+    get_encoding_format,
     get_sms_default_valid_until,
     load_model,
     prefix_generate_users,
@@ -951,7 +952,11 @@ def csvfile_upload(
         if not csvfile:
             csvfile = self.csvfile
         csv_data = csvfile.read()
-        csv_data = csv_data.decode('utf-8') if isinstance(csv_data, bytes) else csv_data
+        csv_data = (
+            csv_data.decode(get_encoding_format(csv_data))
+            if isinstance(csv_data, bytes)
+            else csv_data
+        )
         reader = csv.reader(StringIO(csv_data), delimiter=',')
         self.full_clean()
         self.save()

openwisp_radius/receivers.py

@@ -1,6 +1,7 @@
 """
 Receiver functions for django signals (eg: post_save)
 """
+
 import logging
 
 from celery.exceptions import OperationalError

openwisp_radius/tests/static/test_batch_utf16_file2.csv

@@ -0,0 +1 @@
+44D1FADD7379,cleartext$D0weL6L8,44D1FADD7379@umoja.com,EAPUSER1,USER1

openwisp_radius/tests/test_saml/utils.py

@@ -25,6 +25,6 @@ def add_outstanding_query(self, session_id, came_from):
             came_from,
         )
         self.saml_session.save()
-        self.client.cookies[
-            settings.SESSION_COOKIE_NAME
-        ] = self.saml_session.session_key
+        self.client.cookies[settings.SESSION_COOKIE_NAME] = (
+            self.saml_session.session_key
+        )

openwisp_radius/tests/test_utils.py

@@ -2,7 +2,12 @@
 from django.core.exceptions import ValidationError
 from django.test import override_settings
 
-from ..utils import find_available_username, get_one_time_login_url, validate_csvfile
+from ..utils import (
+    find_available_username,
+    get_encoding_format,
+    get_one_time_login_url,
+    validate_csvfile,
+)
 from . import FileMixin
 from .mixins import BaseTestCase
 
@@ -24,6 +29,12 @@ def test_validate_file_format(self):
             in error.exception.message
         )
 
+    def test_validate_utf16_file_format(self):
+        utf_16_file_1_format_path = self._get_path('static/test_batch_utf16_file1.csv')
+        assert validate_csvfile(open(utf_16_file_1_format_path, 'rb')) is None
+        utf_16_file_2_format_path = self._get_path('static/test_batch_utf16_file2.csv')
+        assert validate_csvfile(open(utf_16_file_2_format_path, 'rb')) is None
+
     def test_validate_csvfile(self):
         invalid_csv_path = self._get_path('static/test_batch_invalid.csv')
         improper_csv_path = self._get_path('static/test_batch_improper.csv')
@@ -34,6 +45,27 @@ def test_validate_csvfile(self):
             validate_csvfile(open(improper_csv_path, 'rt'))
         self.assertTrue('Improper CSV format' in error.exception.message)
 
+    def test_get_encoding_format(self):
+        test_cases = [
+            # UTF-8 encoded data
+            (b'hello world', 'utf-8', "UTF-8 encoded data"),
+            # UTF-16 encoded data with BOM
+            (
+                b'\xff\xfeh\x00e\x00l\x00l\x00o\x00 \x00w\x00o\x00r\x00l\x00d\x00',
+                'utf-16',
+                "UTF-16 encoded data with BOM",
+            ),
+            # Empty data
+            (b'', 'utf-8', "Empty byte data"),
+            # Invalid encoding data
+            (b'\x80\x81\x82', 'utf-8', "Invalid encoding data"),
+        ]
+
+        for data, expected, description in test_cases:
+            with self.subTest(description=description):
+                result = get_encoding_format(data)
+                self.assertEqual(result, expected)
+
     @override_settings(AUTHENTICATION_BACKENDS=[])
     def test_get_one_time_login_url(self):
         login_url = get_one_time_login_url(None, None)

openwisp_radius/utils.py

@@ -4,6 +4,7 @@
 from datetime import timedelta
 from io import BytesIO, StringIO
 
+import chardet
 import swapper
 from django.conf import settings
 from django.contrib.auth import get_user_model
@@ -129,10 +130,29 @@ def find_available_username(username, users_list, prefix=False):
     return tmp
 
 
+def get_encoding_format(csv_data):
+    # Explicit handling for UTF-16 encodings (check for BOM)
+    if csv_data.startswith(b'\xff\xfe') or csv_data.startswith(b'\xfe\xff'):
+        return 'utf-16'
+
+    detected_encoding = chardet.detect(csv_data).get('encoding')
+
+    if detected_encoding == 'ascii':
+        return 'utf-8'
+    if detected_encoding == 'utf-16le':
+        return "utf-16le"
+
+    return detected_encoding or 'utf-8'
+
+
 def validate_csvfile(csvfile):
     csv_data = csvfile.read()
     try:
-        csv_data = csv_data.decode('utf-8') if isinstance(csv_data, bytes) else csv_data
+        csv_data = (
+            csv_data.decode(get_encoding_format(csv_data))
+            if isinstance(csv_data, bytes)
+            else csv_data
+        )
     except UnicodeDecodeError:
         raise ValidationError(
             _(

requirements-test.txt

@@ -13,3 +13,4 @@ openwisp-monitoring @ https://github.com/openwisp/openwisp-monitoring/tarball/ma
 django-redis~=5.4.0
 mock-ssh-server~=0.9.1
 channels_redis~=4.2.1
+chardet~=4.0.0