[16.0][FIX] users_ldap_groups: safe LDAP decode

users_ldap_groups/models/res_company_ldap_operator.py

@@ -1,6 +1,7 @@
 # Copyright 2012-2018 Therp BV <https://therp.nl>
 # Copyright 2018 Brainbean Apps <https://brainbeanapps.com>
 # License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html).
+import base64
 from logging import getLogger
 from string import Template
 
@@ -32,10 +33,27 @@ def _equals(self, ldap_entry, mapping):
 
     def _query(self, ldap_entry, mapping):
         query_string = Template(mapping.value).safe_substitute(
-            {attr: ldap_entry[1][attr][0].decode() for attr in ldap_entry[1]}
+            {
+                attr: self.safe_ldap_decode(ldap_entry[1][attr][0])
+                for attr in ldap_entry[1]
+            }
         )
 
         results = mapping.ldap_id._query(mapping.ldap_id.read()[0], query_string)
         _logger.debug('Performed LDAP query "%s" results: %s', query_string, results)
 
         return bool(results)
+
+    def safe_ldap_decode(self, attr):
+        """Safe LDAP decode; Base64 encode attributes containing binary data.
+        Binary data can be stored in Active Directory, e.g., thumbnailPhoto is
+        stored as binary. As Str.decoce() cannot handle binary, this method
+        Base64 encodes the data in the attribute, instead, if decode fails.
+        Using Base64 should be a suitable fallback, because the use cases of
+        users_ldap_groups do not really require comparing binary attributes.
+        """
+
+        try:
+            return attr.decode()
+        except UnicodeDecodeError:
+            return base64.b64encode(attr)

users_ldap_groups/tests/test_users_ldap_groups.py

@@ -225,3 +225,70 @@ def _test_users_ldap_groups_not_user_type(self):
                 self.env["res.users"].sudo().authenticate(
                     self.env.cr.dbname, "users_ldap_groups-username", "password", {}
                 )
+
+    def test_users_ldap_groups_ldap_returns_binary_data(self):
+        self._create_ldap_config(
+            groups=[
+                {
+                    "ldap_attribute": "name",
+                    "operator": "contains",
+                    "value": "hello3",
+                    "group_id": self.group_system.id,
+                },
+                {
+                    "ldap_attribute": "name",
+                    "operator": "contains",
+                    "value": "hello",
+                    "group_id": self.group_user.id,
+                },
+                {
+                    "ldap_attribute": "name",
+                    "operator": "contains",
+                    "value": "hello2",
+                    "group_id": self.group_contains.id,
+                },
+                {
+                    "ldap_attribute": "name",
+                    "operator": "equals",
+                    "value": "hello",
+                    "group_id": self.group_equals.id,
+                },
+                {
+                    "ldap_attribute": "",
+                    "operator": "query",
+                    "value": "is not run because of patching",
+                    "group_id": self.group_query.id,
+                },
+            ],
+            only_ldap_groups=True,
+        )
+        with mock.patch(
+            _company_ldap_class + "._connect",
+            return_value=FakeLdapConnection(
+                {
+                    "dc=users_ldap_groups,dc=example,dc=com": {
+                        "cn": [b"User Name"],
+                        "name": [b"hello", b"hello2"],
+                        "thumbnailPhoto": [
+                            b"GIF89a\x01\x00\x01\x00\x00\xff\x00,"
+                            b"\x00\x00\x00\x00\x01\x00\x01\x00\x00\x02\x00;"
+                        ],
+                    }
+                }
+            ),
+        ), mock_cursor(self.cr):
+            user_id = (
+                self.env["res.users"]
+                .sudo()
+                .authenticate(
+                    self.env.cr.dbname, "users_ldap_groups-username", "password", {}
+                )
+            )
+        # this asserts group mappings from demo data
+        user = self.env["res.users"].sudo().browse(user_id)
+        groups = user.groups_id
+        self.assertIn(self.group_contains, groups)
+        self.assertIn(self.group_user, groups)
+        self.assertNotIn(self.group_equals, groups)
+        self.assertIn(self.group_query, groups)
+        self.assertNotIn(self.group_system, groups)